1use std::collections::{BTreeMap, BTreeSet};
2use std::sync::{Arc, OnceLock, RwLock};
3
4use serde::{Deserialize, Deserializer, Serialize, Serializer};
5use serde_json::Value as JsonValue;
6use time::OffsetDateTime;
7use uuid::Uuid;
8
9#[cfg(test)]
10use crate::redact::REDACTED_HEADER_VALUE;
11use crate::triggers::test_util::clock;
12
13fn serialize_optional_bytes_b64<S>(
14 value: &Option<Vec<u8>>,
15 serializer: S,
16) -> Result<S::Ok, S::Error>
17where
18 S: Serializer,
19{
20 use base64::Engine;
21
22 match value {
23 Some(bytes) => {
24 serializer.serialize_some(&base64::engine::general_purpose::STANDARD.encode(bytes))
25 }
26 None => serializer.serialize_none(),
27 }
28}
29
30fn deserialize_optional_bytes_b64<'de, D>(deserializer: D) -> Result<Option<Vec<u8>>, D::Error>
31where
32 D: Deserializer<'de>,
33{
34 use base64::Engine;
35
36 let encoded = Option::<String>::deserialize(deserializer)?;
37 encoded
38 .map(|text| {
39 base64::engine::general_purpose::STANDARD
40 .decode(text.as_bytes())
41 .map_err(serde::de::Error::custom)
42 })
43 .transpose()
44}
45
46#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
47#[serde(transparent)]
48pub struct TriggerEventId(pub String);
49
50impl TriggerEventId {
51 pub fn new() -> Self {
52 Self(format!("trigger_evt_{}", Uuid::now_v7()))
53 }
54}
55
56impl Default for TriggerEventId {
57 fn default() -> Self {
58 Self::new()
59 }
60}
61
62#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
63#[serde(transparent)]
64pub struct ProviderId(pub String);
65
66impl ProviderId {
67 pub fn new(value: impl Into<String>) -> Self {
68 Self(value.into())
69 }
70
71 pub fn as_str(&self) -> &str {
72 self.0.as_str()
73 }
74}
75
76impl From<&str> for ProviderId {
77 fn from(value: &str) -> Self {
78 Self::new(value)
79 }
80}
81
82impl From<String> for ProviderId {
83 fn from(value: String) -> Self {
84 Self::new(value)
85 }
86}
87
88#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
89#[serde(transparent)]
90pub struct TraceId(pub String);
91
92impl TraceId {
93 pub fn new() -> Self {
94 Self(format!("trace_{}", Uuid::now_v7()))
95 }
96}
97
98impl Default for TraceId {
99 fn default() -> Self {
100 Self::new()
101 }
102}
103
104#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
105#[serde(transparent)]
106pub struct TenantId(pub String);
107
108impl TenantId {
109 pub fn new(value: impl Into<String>) -> Self {
110 Self(value.into())
111 }
112}
113
114#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
115#[serde(tag = "state", rename_all = "snake_case")]
116pub enum SignatureStatus {
117 Verified,
118 Unsigned,
119 Failed { reason: String },
120}
121
122#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
123pub struct GitHubEventCommon {
124 pub event: String,
125 pub action: Option<String>,
126 pub delivery_id: Option<String>,
127 pub installation_id: Option<i64>,
128 #[serde(default, skip_serializing_if = "Option::is_none")]
129 pub topic: Option<String>,
130 #[serde(default, skip_serializing_if = "Option::is_none")]
131 pub repository: Option<JsonValue>,
132 #[serde(default, skip_serializing_if = "Option::is_none")]
133 pub repo: Option<JsonValue>,
134 pub raw: JsonValue,
135}
136
137#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
138pub struct GitHubIssuesEventPayload {
139 #[serde(flatten)]
140 pub common: GitHubEventCommon,
141 pub issue: JsonValue,
142}
143
144#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
145pub struct GitHubPullRequestEventPayload {
146 #[serde(flatten)]
147 pub common: GitHubEventCommon,
148 pub pull_request: JsonValue,
149}
150
151#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
152pub struct GitHubIssueCommentEventPayload {
153 #[serde(flatten)]
154 pub common: GitHubEventCommon,
155 pub issue: JsonValue,
156 pub comment: JsonValue,
157}
158
159#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
160pub struct GitHubPullRequestReviewEventPayload {
161 #[serde(flatten)]
162 pub common: GitHubEventCommon,
163 pub pull_request: JsonValue,
164 pub review: JsonValue,
165}
166
167#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
168pub struct GitHubPushEventPayload {
169 #[serde(flatten)]
170 pub common: GitHubEventCommon,
171 #[serde(default)]
172 pub commits: Vec<JsonValue>,
173 pub distinct_size: Option<i64>,
174}
175
176#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
177pub struct GitHubWorkflowRunEventPayload {
178 #[serde(flatten)]
179 pub common: GitHubEventCommon,
180 pub workflow_run: JsonValue,
181}
182
183#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
184pub struct GitHubDeploymentStatusEventPayload {
185 #[serde(flatten)]
186 pub common: GitHubEventCommon,
187 pub deployment_status: JsonValue,
188 pub deployment: JsonValue,
189}
190
191#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
192pub struct GitHubCheckRunEventPayload {
193 #[serde(flatten)]
194 pub common: GitHubEventCommon,
195 pub check_run: JsonValue,
196}
197
198#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
199pub struct GitHubCheckSuiteEventPayload {
200 #[serde(flatten)]
201 pub common: GitHubEventCommon,
202 pub check_suite: JsonValue,
203 #[serde(default, skip_serializing_if = "Option::is_none")]
204 pub check_suite_id: Option<i64>,
205 #[serde(default, skip_serializing_if = "Option::is_none")]
206 pub pull_request_number: Option<i64>,
207 #[serde(default, skip_serializing_if = "Option::is_none")]
208 pub head_sha: Option<String>,
209 #[serde(default, skip_serializing_if = "Option::is_none")]
210 pub head_ref: Option<String>,
211 #[serde(default, skip_serializing_if = "Option::is_none")]
212 pub base_ref: Option<String>,
213 #[serde(default, skip_serializing_if = "Option::is_none")]
214 pub status: Option<String>,
215 #[serde(default, skip_serializing_if = "Option::is_none")]
216 pub conclusion: Option<String>,
217}
218
219#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
220pub struct GitHubStatusEventPayload {
221 #[serde(flatten)]
222 pub common: GitHubEventCommon,
223 #[serde(default, skip_serializing_if = "Option::is_none")]
224 pub commit_status: Option<JsonValue>,
225 #[serde(default, skip_serializing_if = "Option::is_none")]
226 pub status_id: Option<i64>,
227 #[serde(default, skip_serializing_if = "Option::is_none")]
228 pub head_sha: Option<String>,
229 #[serde(default, skip_serializing_if = "Option::is_none")]
230 pub head_ref: Option<String>,
231 #[serde(default, skip_serializing_if = "Option::is_none")]
232 pub base_ref: Option<String>,
233 #[serde(default, skip_serializing_if = "Option::is_none")]
234 pub state: Option<String>,
235 #[serde(default, skip_serializing_if = "Option::is_none")]
236 pub context: Option<String>,
237 #[serde(default, skip_serializing_if = "Option::is_none")]
238 pub target_url: Option<String>,
239}
240
241#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
242pub struct GitHubMergeGroupEventPayload {
243 #[serde(flatten)]
244 pub common: GitHubEventCommon,
245 pub merge_group: JsonValue,
246 #[serde(default, skip_serializing_if = "Option::is_none")]
247 pub merge_group_id: Option<JsonValue>,
248 #[serde(default, skip_serializing_if = "Option::is_none")]
249 pub head_sha: Option<String>,
250 #[serde(default, skip_serializing_if = "Option::is_none")]
251 pub head_ref: Option<String>,
252 #[serde(default, skip_serializing_if = "Option::is_none")]
253 pub base_sha: Option<String>,
254 #[serde(default, skip_serializing_if = "Option::is_none")]
255 pub base_ref: Option<String>,
256 #[serde(default)]
257 pub pull_requests: Vec<JsonValue>,
258 #[serde(default)]
259 pub pull_request_numbers: Vec<i64>,
260}
261
262#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
263pub struct GitHubInstallationEventPayload {
264 #[serde(flatten)]
265 pub common: GitHubEventCommon,
266 #[serde(default, skip_serializing_if = "Option::is_none")]
267 pub installation: Option<JsonValue>,
268 #[serde(default, skip_serializing_if = "Option::is_none")]
269 pub account: Option<JsonValue>,
270 #[serde(default, skip_serializing_if = "Option::is_none")]
271 pub installation_state: Option<String>,
272 #[serde(default, skip_serializing_if = "Option::is_none")]
273 pub suspended: Option<bool>,
274 #[serde(default, skip_serializing_if = "Option::is_none")]
275 pub revoked: Option<bool>,
276 #[serde(default)]
277 pub repositories: Vec<JsonValue>,
278}
279
280#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
281pub struct GitHubInstallationRepositoriesEventPayload {
282 #[serde(flatten)]
283 pub common: GitHubEventCommon,
284 #[serde(default, skip_serializing_if = "Option::is_none")]
285 pub installation: Option<JsonValue>,
286 #[serde(default, skip_serializing_if = "Option::is_none")]
287 pub account: Option<JsonValue>,
288 #[serde(default, skip_serializing_if = "Option::is_none")]
289 pub installation_state: Option<String>,
290 #[serde(default, skip_serializing_if = "Option::is_none")]
291 pub suspended: Option<bool>,
292 #[serde(default, skip_serializing_if = "Option::is_none")]
293 pub revoked: Option<bool>,
294 #[serde(default, skip_serializing_if = "Option::is_none")]
295 pub repository_selection: Option<String>,
296 #[serde(default)]
297 pub repositories_added: Vec<JsonValue>,
298 #[serde(default)]
299 pub repositories_removed: Vec<JsonValue>,
300}
301
302#[derive(Clone, Debug, PartialEq, Serialize)]
303#[serde(untagged)]
304pub enum GitHubEventPayload {
305 Issues(GitHubIssuesEventPayload),
306 PullRequest(GitHubPullRequestEventPayload),
307 IssueComment(GitHubIssueCommentEventPayload),
308 PullRequestReview(GitHubPullRequestReviewEventPayload),
309 Push(GitHubPushEventPayload),
310 WorkflowRun(GitHubWorkflowRunEventPayload),
311 DeploymentStatus(GitHubDeploymentStatusEventPayload),
312 CheckRun(GitHubCheckRunEventPayload),
313 CheckSuite(GitHubCheckSuiteEventPayload),
314 Status(GitHubStatusEventPayload),
315 MergeGroup(GitHubMergeGroupEventPayload),
316 Installation(GitHubInstallationEventPayload),
317 InstallationRepositories(GitHubInstallationRepositoriesEventPayload),
318 Other(GitHubEventCommon),
319}
320
321impl<'de> Deserialize<'de> for GitHubEventPayload {
326 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
327 where
328 D: Deserializer<'de>,
329 {
330 let value = JsonValue::deserialize(deserializer)?;
331 let kind = value
332 .get("event")
333 .and_then(JsonValue::as_str)
334 .unwrap_or("")
335 .to_string();
336 let from_value = |v: JsonValue| -> Result<GitHubEventPayload, D::Error> {
337 let payload = match kind.as_str() {
338 "issues" => GitHubEventPayload::Issues(
339 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
340 ),
341 "pull_request" => GitHubEventPayload::PullRequest(
342 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
343 ),
344 "issue_comment" => GitHubEventPayload::IssueComment(
345 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
346 ),
347 "pull_request_review" => GitHubEventPayload::PullRequestReview(
348 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
349 ),
350 "push" => GitHubEventPayload::Push(
351 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
352 ),
353 "workflow_run" => GitHubEventPayload::WorkflowRun(
354 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
355 ),
356 "deployment_status" => GitHubEventPayload::DeploymentStatus(
357 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
358 ),
359 "check_run" => GitHubEventPayload::CheckRun(
360 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
361 ),
362 "check_suite" => GitHubEventPayload::CheckSuite(
363 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
364 ),
365 "status" => GitHubEventPayload::Status(
366 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
367 ),
368 "merge_group" => GitHubEventPayload::MergeGroup(
369 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
370 ),
371 "installation" => GitHubEventPayload::Installation(
372 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
373 ),
374 "installation_repositories" => GitHubEventPayload::InstallationRepositories(
375 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
376 ),
377 _ => GitHubEventPayload::Other(
378 serde_json::from_value(v).map_err(serde::de::Error::custom)?,
379 ),
380 };
381 Ok(payload)
382 };
383 from_value(value)
384 }
385}
386
387#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
388pub struct SlackEventCommon {
389 pub event: String,
390 pub event_id: Option<String>,
391 pub api_app_id: Option<String>,
392 pub team_id: Option<String>,
393 pub channel_id: Option<String>,
394 pub user_id: Option<String>,
395 pub event_ts: Option<String>,
396 pub raw: JsonValue,
397}
398
399#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
400pub struct SlackMessageEventPayload {
401 #[serde(flatten)]
402 pub common: SlackEventCommon,
403 pub subtype: Option<String>,
404 pub channel_type: Option<String>,
405 pub channel: Option<String>,
406 pub user: Option<String>,
407 pub text: Option<String>,
408 pub ts: Option<String>,
409 pub thread_ts: Option<String>,
410}
411
412#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
413pub struct SlackAppMentionEventPayload {
414 #[serde(flatten)]
415 pub common: SlackEventCommon,
416 pub channel: Option<String>,
417 pub user: Option<String>,
418 pub text: Option<String>,
419 pub ts: Option<String>,
420 pub thread_ts: Option<String>,
421}
422
423#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
424pub struct SlackReactionAddedEventPayload {
425 #[serde(flatten)]
426 pub common: SlackEventCommon,
427 pub reaction: Option<String>,
428 pub item_user: Option<String>,
429 pub item: JsonValue,
430}
431
432#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
433pub struct SlackAppHomeOpenedEventPayload {
434 #[serde(flatten)]
435 pub common: SlackEventCommon,
436 pub user: Option<String>,
437 pub channel: Option<String>,
438 pub tab: Option<String>,
439 pub view: JsonValue,
440}
441
442#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
443pub struct SlackAssistantThreadStartedEventPayload {
444 #[serde(flatten)]
445 pub common: SlackEventCommon,
446 pub assistant_thread: JsonValue,
447 pub thread_ts: Option<String>,
448 pub context: JsonValue,
449}
450
451#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
452#[serde(untagged)]
453pub enum SlackEventPayload {
454 Message(SlackMessageEventPayload),
455 AppMention(SlackAppMentionEventPayload),
456 ReactionAdded(SlackReactionAddedEventPayload),
457 AppHomeOpened(SlackAppHomeOpenedEventPayload),
458 AssistantThreadStarted(SlackAssistantThreadStartedEventPayload),
459 Other(SlackEventCommon),
460}
461
462#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
463pub struct LinearEventCommon {
464 pub event: String,
465 pub action: Option<String>,
466 pub delivery_id: Option<String>,
467 pub organization_id: Option<String>,
468 pub webhook_timestamp: Option<i64>,
469 pub webhook_id: Option<String>,
470 pub url: Option<String>,
471 pub created_at: Option<String>,
472 pub actor: JsonValue,
473 pub raw: JsonValue,
474}
475
476#[derive(Clone, Debug, PartialEq)]
477pub enum LinearIssueChange {
478 Title { previous: Option<String> },
479 Description { previous: Option<String> },
480 Priority { previous: Option<i64> },
481 Estimate { previous: Option<i64> },
482 StateId { previous: Option<String> },
483 TeamId { previous: Option<String> },
484 AssigneeId { previous: Option<String> },
485 ProjectId { previous: Option<String> },
486 CycleId { previous: Option<String> },
487 DueDate { previous: Option<String> },
488 ParentId { previous: Option<String> },
489 SortOrder { previous: Option<f64> },
490 LabelIds { previous: Vec<String> },
491 CompletedAt { previous: Option<String> },
492 Other { field: String, previous: JsonValue },
493}
494
495impl Serialize for LinearIssueChange {
496 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
497 where
498 S: Serializer,
499 {
500 let value = match self {
501 Self::Title { previous } => {
502 serde_json::json!({ "field_name": "title", "previous": previous })
503 }
504 Self::Description { previous } => {
505 serde_json::json!({ "field_name": "description", "previous": previous })
506 }
507 Self::Priority { previous } => {
508 serde_json::json!({ "field_name": "priority", "previous": previous })
509 }
510 Self::Estimate { previous } => {
511 serde_json::json!({ "field_name": "estimate", "previous": previous })
512 }
513 Self::StateId { previous } => {
514 serde_json::json!({ "field_name": "state_id", "previous": previous })
515 }
516 Self::TeamId { previous } => {
517 serde_json::json!({ "field_name": "team_id", "previous": previous })
518 }
519 Self::AssigneeId { previous } => {
520 serde_json::json!({ "field_name": "assignee_id", "previous": previous })
521 }
522 Self::ProjectId { previous } => {
523 serde_json::json!({ "field_name": "project_id", "previous": previous })
524 }
525 Self::CycleId { previous } => {
526 serde_json::json!({ "field_name": "cycle_id", "previous": previous })
527 }
528 Self::DueDate { previous } => {
529 serde_json::json!({ "field_name": "due_date", "previous": previous })
530 }
531 Self::ParentId { previous } => {
532 serde_json::json!({ "field_name": "parent_id", "previous": previous })
533 }
534 Self::SortOrder { previous } => {
535 serde_json::json!({ "field_name": "sort_order", "previous": previous })
536 }
537 Self::LabelIds { previous } => {
538 serde_json::json!({ "field_name": "label_ids", "previous": previous })
539 }
540 Self::CompletedAt { previous } => {
541 serde_json::json!({ "field_name": "completed_at", "previous": previous })
542 }
543 Self::Other { field, previous } => {
544 serde_json::json!({ "field_name": "other", "field": field, "previous": previous })
545 }
546 };
547 value.serialize(serializer)
548 }
549}
550
551impl<'de> Deserialize<'de> for LinearIssueChange {
552 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
553 where
554 D: Deserializer<'de>,
555 {
556 let value = JsonValue::deserialize(deserializer)?;
557 let field_name = value
558 .get("field_name")
559 .and_then(JsonValue::as_str)
560 .ok_or_else(|| serde::de::Error::custom("linear issue change missing field_name"))?;
561 let previous = value.get("previous").cloned().unwrap_or(JsonValue::Null);
562 Ok(match field_name {
563 "title" => Self::Title {
564 previous: previous.as_str().map(ToString::to_string),
565 },
566 "description" => Self::Description {
567 previous: previous.as_str().map(ToString::to_string),
568 },
569 "priority" => Self::Priority {
570 previous: parse_json_i64ish(&previous),
571 },
572 "estimate" => Self::Estimate {
573 previous: parse_json_i64ish(&previous),
574 },
575 "state_id" => Self::StateId {
576 previous: previous.as_str().map(ToString::to_string),
577 },
578 "team_id" => Self::TeamId {
579 previous: previous.as_str().map(ToString::to_string),
580 },
581 "assignee_id" => Self::AssigneeId {
582 previous: previous.as_str().map(ToString::to_string),
583 },
584 "project_id" => Self::ProjectId {
585 previous: previous.as_str().map(ToString::to_string),
586 },
587 "cycle_id" => Self::CycleId {
588 previous: previous.as_str().map(ToString::to_string),
589 },
590 "due_date" => Self::DueDate {
591 previous: previous.as_str().map(ToString::to_string),
592 },
593 "parent_id" => Self::ParentId {
594 previous: previous.as_str().map(ToString::to_string),
595 },
596 "sort_order" => Self::SortOrder {
597 previous: previous.as_f64(),
598 },
599 "label_ids" => Self::LabelIds {
600 previous: parse_string_array(&previous),
601 },
602 "completed_at" => Self::CompletedAt {
603 previous: previous.as_str().map(ToString::to_string),
604 },
605 "other" => Self::Other {
606 field: value
607 .get("field")
608 .and_then(JsonValue::as_str)
609 .map(ToString::to_string)
610 .unwrap_or_else(|| "unknown".to_string()),
611 previous,
612 },
613 other => Self::Other {
614 field: other.to_string(),
615 previous,
616 },
617 })
618 }
619}
620
621#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
622pub struct LinearIssueEventPayload {
623 #[serde(flatten)]
624 pub common: LinearEventCommon,
625 pub issue: JsonValue,
626 #[serde(default)]
627 pub changes: Vec<LinearIssueChange>,
628}
629
630#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
631pub struct LinearIssueCommentEventPayload {
632 #[serde(flatten)]
633 pub common: LinearEventCommon,
634 pub comment: JsonValue,
635}
636
637#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
638pub struct LinearIssueLabelEventPayload {
639 #[serde(flatten)]
640 pub common: LinearEventCommon,
641 pub label: JsonValue,
642}
643
644#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
645pub struct LinearProjectEventPayload {
646 #[serde(flatten)]
647 pub common: LinearEventCommon,
648 pub project: JsonValue,
649}
650
651#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
652pub struct LinearCycleEventPayload {
653 #[serde(flatten)]
654 pub common: LinearEventCommon,
655 pub cycle: JsonValue,
656}
657
658#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
659pub struct LinearCustomerEventPayload {
660 #[serde(flatten)]
661 pub common: LinearEventCommon,
662 pub customer: JsonValue,
663}
664
665#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
666pub struct LinearCustomerRequestEventPayload {
667 #[serde(flatten)]
668 pub common: LinearEventCommon,
669 pub customer_request: JsonValue,
670}
671
672#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
673#[serde(untagged)]
674pub enum LinearEventPayload {
675 Issue(LinearIssueEventPayload),
676 IssueComment(LinearIssueCommentEventPayload),
677 IssueLabel(LinearIssueLabelEventPayload),
678 Project(LinearProjectEventPayload),
679 Cycle(LinearCycleEventPayload),
680 Customer(LinearCustomerEventPayload),
681 CustomerRequest(LinearCustomerRequestEventPayload),
682 Other(LinearEventCommon),
683}
684
685#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
686pub struct NotionPolledChangeEvent {
687 pub resource: String,
688 pub source_id: String,
689 pub entity_id: String,
690 pub high_water_mark: String,
691 pub before: Option<JsonValue>,
692 pub after: JsonValue,
693}
694
695#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
696pub struct NotionEventPayload {
697 pub event: String,
698 pub workspace_id: Option<String>,
699 pub request_id: Option<String>,
700 pub subscription_id: Option<String>,
701 pub integration_id: Option<String>,
702 pub attempt_number: Option<u32>,
703 pub entity_id: Option<String>,
704 pub entity_type: Option<String>,
705 pub api_version: Option<String>,
706 pub verification_token: Option<String>,
707 pub polled: Option<NotionPolledChangeEvent>,
708 pub raw: JsonValue,
709}
710
711#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
712pub struct CronEventPayload {
713 pub cron_id: Option<String>,
714 pub schedule: Option<String>,
715 #[serde(with = "time::serde::rfc3339")]
716 pub tick_at: OffsetDateTime,
717 pub raw: JsonValue,
718}
719
720#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
721pub struct GenericWebhookPayload {
722 pub source: Option<String>,
723 pub content_type: Option<String>,
724 pub raw: JsonValue,
725}
726
727#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
728pub struct A2aPushPayload {
729 pub task_id: Option<String>,
730 pub task_state: Option<String>,
731 pub artifact: Option<JsonValue>,
732 pub sender: Option<String>,
733 pub raw: JsonValue,
734 pub kind: String,
735}
736
737#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
738pub struct StreamEventPayload {
739 pub event: String,
740 pub source: Option<String>,
741 pub stream: Option<String>,
742 pub partition: Option<String>,
743 pub offset: Option<String>,
744 pub key: Option<String>,
745 pub timestamp: Option<String>,
746 #[serde(default)]
747 pub headers: BTreeMap<String, String>,
748 pub raw: JsonValue,
749}
750
751#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
752pub struct ExtensionProviderPayload {
753 pub provider: String,
754 pub schema_name: String,
755 pub raw: JsonValue,
756}
757
758#[allow(clippy::large_enum_variant)]
759#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
760#[serde(untagged)]
761pub enum ProviderPayload {
762 Known(KnownProviderPayload),
763 Extension(ExtensionProviderPayload),
764}
765
766impl ProviderPayload {
767 pub fn provider(&self) -> &str {
768 match self {
769 Self::Known(known) => known.provider(),
770 Self::Extension(payload) => payload.provider.as_str(),
771 }
772 }
773
774 pub fn normalize(
775 provider: &ProviderId,
776 kind: &str,
777 headers: &BTreeMap<String, String>,
778 raw: JsonValue,
779 ) -> Result<Self, ProviderCatalogError> {
780 provider_catalog()
781 .read()
782 .expect("provider catalog poisoned")
783 .normalize(provider, kind, headers, raw)
784 }
785}
786
787#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
788#[serde(tag = "provider")]
789pub enum KnownProviderPayload {
790 #[serde(rename = "github")]
791 GitHub(GitHubEventPayload),
792 #[serde(rename = "slack")]
793 Slack(Box<SlackEventPayload>),
794 #[serde(rename = "linear")]
795 Linear(LinearEventPayload),
796 #[serde(rename = "notion")]
797 Notion(Box<NotionEventPayload>),
798 #[serde(rename = "cron")]
799 Cron(CronEventPayload),
800 #[serde(rename = "webhook")]
801 Webhook(GenericWebhookPayload),
802 #[serde(rename = "a2a-push")]
803 A2aPush(A2aPushPayload),
804 #[serde(rename = "kafka")]
805 Kafka(StreamEventPayload),
806 #[serde(rename = "nats")]
807 Nats(StreamEventPayload),
808 #[serde(rename = "pulsar")]
809 Pulsar(StreamEventPayload),
810 #[serde(rename = "postgres-cdc")]
811 PostgresCdc(StreamEventPayload),
812 #[serde(rename = "email")]
813 Email(StreamEventPayload),
814 #[serde(rename = "websocket")]
815 Websocket(StreamEventPayload),
816}
817
818impl KnownProviderPayload {
819 pub fn provider(&self) -> &str {
820 match self {
821 Self::GitHub(_) => "github",
822 Self::Slack(_) => "slack",
823 Self::Linear(_) => "linear",
824 Self::Notion(_) => "notion",
825 Self::Cron(_) => "cron",
826 Self::Webhook(_) => "webhook",
827 Self::A2aPush(_) => "a2a-push",
828 Self::Kafka(_) => "kafka",
829 Self::Nats(_) => "nats",
830 Self::Pulsar(_) => "pulsar",
831 Self::PostgresCdc(_) => "postgres-cdc",
832 Self::Email(_) => "email",
833 Self::Websocket(_) => "websocket",
834 }
835 }
836}
837
838#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
839pub struct TriggerEvent {
840 pub id: TriggerEventId,
841 pub provider: ProviderId,
842 pub kind: String,
843 #[serde(with = "time::serde::rfc3339")]
844 pub received_at: OffsetDateTime,
845 #[serde(with = "time::serde::rfc3339::option")]
846 pub occurred_at: Option<OffsetDateTime>,
847 pub dedupe_key: String,
848 pub trace_id: TraceId,
849 pub tenant_id: Option<TenantId>,
850 pub headers: BTreeMap<String, String>,
851 #[serde(default, skip_serializing_if = "Option::is_none")]
852 pub batch: Option<Vec<JsonValue>>,
853 #[serde(
854 default,
855 skip_serializing_if = "Option::is_none",
856 serialize_with = "serialize_optional_bytes_b64",
857 deserialize_with = "deserialize_optional_bytes_b64"
858 )]
859 pub raw_body: Option<Vec<u8>>,
860 pub provider_payload: ProviderPayload,
861 pub signature_status: SignatureStatus,
862 #[serde(skip)]
863 pub dedupe_claimed: bool,
864}
865
866impl TriggerEvent {
867 #[allow(clippy::too_many_arguments)]
868 pub fn new(
869 provider: ProviderId,
870 kind: impl Into<String>,
871 occurred_at: Option<OffsetDateTime>,
872 dedupe_key: impl Into<String>,
873 tenant_id: Option<TenantId>,
874 headers: BTreeMap<String, String>,
875 provider_payload: ProviderPayload,
876 signature_status: SignatureStatus,
877 ) -> Self {
878 Self {
879 id: TriggerEventId::new(),
880 provider,
881 kind: kind.into(),
882 received_at: clock::now_utc(),
883 occurred_at,
884 dedupe_key: dedupe_key.into(),
885 trace_id: TraceId::new(),
886 tenant_id,
887 headers,
888 batch: None,
889 raw_body: None,
890 provider_payload,
891 signature_status,
892 dedupe_claimed: false,
893 }
894 }
895
896 pub fn dedupe_claimed(&self) -> bool {
897 self.dedupe_claimed
898 }
899
900 pub fn mark_dedupe_claimed(&mut self) {
901 self.dedupe_claimed = true;
902 }
903}
904
905pub type HeaderRedactionPolicy = crate::redact::RedactionPolicy;
912
913pub fn redact_headers(
914 headers: &BTreeMap<String, String>,
915 policy: &HeaderRedactionPolicy,
916) -> BTreeMap<String, String> {
917 policy.redact_headers(headers)
918}
919
920#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
921pub struct ProviderSecretRequirement {
922 pub name: String,
923 pub required: bool,
924 pub namespace: String,
925}
926
927#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
928pub struct ProviderOutboundMethod {
929 pub name: String,
930}
931
932#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
933#[serde(tag = "kind", rename_all = "snake_case")]
934pub enum SignatureVerificationMetadata {
935 #[default]
936 None,
937 Hmac {
938 variant: String,
939 raw_body: bool,
940 signature_header: String,
941 timestamp_header: Option<String>,
942 id_header: Option<String>,
943 default_tolerance_secs: Option<i64>,
944 digest: String,
945 encoding: String,
946 },
947}
948
949#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
950#[serde(tag = "kind", rename_all = "snake_case")]
951pub enum ProviderRuntimeMetadata {
952 Builtin {
953 connector: String,
954 default_signature_variant: Option<String>,
955 },
956 #[default]
957 Placeholder,
958}
959
960#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
961pub struct ProviderMetadata {
962 pub provider: String,
963 #[serde(default)]
964 pub kinds: Vec<String>,
965 pub schema_name: String,
966 #[serde(default)]
967 pub outbound_methods: Vec<ProviderOutboundMethod>,
968 #[serde(default)]
969 pub secret_requirements: Vec<ProviderSecretRequirement>,
970 #[serde(default)]
971 pub signature_verification: SignatureVerificationMetadata,
972 #[serde(default)]
973 pub runtime: ProviderRuntimeMetadata,
974}
975
976impl ProviderMetadata {
977 pub fn supports_kind(&self, kind: &str) -> bool {
978 self.kinds.iter().any(|candidate| candidate == kind)
979 }
980
981 pub fn required_secret_names(&self) -> impl Iterator<Item = &str> {
982 self.secret_requirements
983 .iter()
984 .filter(|requirement| requirement.required)
985 .map(|requirement| requirement.name.as_str())
986 }
987}
988
989pub trait ProviderSchema: Send + Sync {
990 fn provider_id(&self) -> &str;
991 fn harn_schema_name(&self) -> &str;
992 fn metadata(&self) -> ProviderMetadata {
993 ProviderMetadata {
994 provider: self.provider_id().to_string(),
995 schema_name: self.harn_schema_name().to_string(),
996 ..ProviderMetadata::default()
997 }
998 }
999 fn normalize(
1000 &self,
1001 kind: &str,
1002 headers: &BTreeMap<String, String>,
1003 raw: JsonValue,
1004 ) -> Result<ProviderPayload, ProviderCatalogError>;
1005}
1006
1007#[derive(Clone, Debug, PartialEq, Eq)]
1008pub enum ProviderCatalogError {
1009 DuplicateProvider(String),
1010 UnknownProvider(String),
1011}
1012
1013impl std::fmt::Display for ProviderCatalogError {
1014 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1015 match self {
1016 Self::DuplicateProvider(provider) => {
1017 write!(f, "provider `{provider}` is already registered")
1018 }
1019 Self::UnknownProvider(provider) => write!(f, "provider `{provider}` is not registered"),
1020 }
1021 }
1022}
1023
1024impl std::error::Error for ProviderCatalogError {}
1025
1026#[derive(Clone, Default)]
1027pub struct ProviderCatalog {
1028 providers: BTreeMap<String, Arc<dyn ProviderSchema>>,
1029}
1030
1031impl ProviderCatalog {
1032 pub fn with_defaults() -> Self {
1033 let mut catalog = Self::default();
1034 for schema in default_provider_schemas() {
1035 catalog
1036 .register(schema)
1037 .expect("default providers must register cleanly");
1038 }
1039 catalog
1040 }
1041
1042 pub fn with_defaults_and(
1043 schemas: Vec<Arc<dyn ProviderSchema>>,
1044 ) -> Result<Self, ProviderCatalogError> {
1045 let mut catalog = Self::with_defaults();
1046 let builtin_providers: BTreeSet<String> = catalog.schema_names().into_keys().collect();
1047 for schema in schemas {
1048 if builtin_providers.contains(schema.provider_id()) {
1049 continue;
1050 }
1051 catalog.register(schema)?;
1052 }
1053 Ok(catalog)
1054 }
1055
1056 pub fn register(
1057 &mut self,
1058 schema: Arc<dyn ProviderSchema>,
1059 ) -> Result<(), ProviderCatalogError> {
1060 let provider = schema.provider_id().to_string();
1061 if self.providers.contains_key(provider.as_str()) {
1062 return Err(ProviderCatalogError::DuplicateProvider(provider));
1063 }
1064 self.providers.insert(provider, schema);
1065 Ok(())
1066 }
1067
1068 pub fn normalize(
1069 &self,
1070 provider: &ProviderId,
1071 kind: &str,
1072 headers: &BTreeMap<String, String>,
1073 raw: JsonValue,
1074 ) -> Result<ProviderPayload, ProviderCatalogError> {
1075 let schema = self
1076 .providers
1077 .get(provider.as_str())
1078 .ok_or_else(|| ProviderCatalogError::UnknownProvider(provider.0.clone()))?;
1079 schema.normalize(kind, headers, raw)
1080 }
1081
1082 pub fn schema_names(&self) -> BTreeMap<String, String> {
1083 self.providers
1084 .iter()
1085 .map(|(provider, schema)| (provider.clone(), schema.harn_schema_name().to_string()))
1086 .collect()
1087 }
1088
1089 pub fn entries(&self) -> Vec<ProviderMetadata> {
1090 self.providers
1091 .values()
1092 .map(|schema| schema.metadata())
1093 .collect()
1094 }
1095
1096 pub fn metadata_for(&self, provider: &str) -> Option<ProviderMetadata> {
1097 self.providers.get(provider).map(|schema| schema.metadata())
1098 }
1099}
1100
1101pub fn register_provider_schema(
1102 schema: Arc<dyn ProviderSchema>,
1103) -> Result<(), ProviderCatalogError> {
1104 provider_catalog()
1105 .write()
1106 .expect("provider catalog poisoned")
1107 .register(schema)
1108}
1109
1110pub fn reset_provider_catalog() {
1111 *provider_catalog()
1112 .write()
1113 .expect("provider catalog poisoned") = ProviderCatalog::with_defaults();
1114}
1115
1116pub fn reset_provider_catalog_with(
1117 schemas: Vec<Arc<dyn ProviderSchema>>,
1118) -> Result<(), ProviderCatalogError> {
1119 let catalog = ProviderCatalog::with_defaults_and(schemas)?;
1120 install_provider_catalog(catalog);
1121 Ok(())
1122}
1123
1124pub fn install_provider_catalog(catalog: ProviderCatalog) {
1125 *provider_catalog()
1126 .write()
1127 .expect("provider catalog poisoned") = catalog;
1128}
1129
1130pub fn registered_provider_schema_names() -> BTreeMap<String, String> {
1131 provider_catalog()
1132 .read()
1133 .expect("provider catalog poisoned")
1134 .schema_names()
1135}
1136
1137pub fn registered_provider_metadata() -> Vec<ProviderMetadata> {
1138 provider_catalog()
1139 .read()
1140 .expect("provider catalog poisoned")
1141 .entries()
1142}
1143
1144pub fn provider_metadata(provider: &str) -> Option<ProviderMetadata> {
1145 provider_catalog()
1146 .read()
1147 .expect("provider catalog poisoned")
1148 .metadata_for(provider)
1149}
1150
1151fn provider_catalog() -> &'static RwLock<ProviderCatalog> {
1152 static PROVIDER_CATALOG: OnceLock<RwLock<ProviderCatalog>> = OnceLock::new();
1153 PROVIDER_CATALOG.get_or_init(|| RwLock::new(ProviderCatalog::with_defaults()))
1154}
1155
1156struct BuiltinProviderSchema {
1157 provider_id: &'static str,
1158 harn_schema_name: &'static str,
1159 metadata: ProviderMetadata,
1160 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
1161}
1162
1163impl ProviderSchema for BuiltinProviderSchema {
1164 fn provider_id(&self) -> &str {
1165 self.provider_id
1166 }
1167
1168 fn harn_schema_name(&self) -> &str {
1169 self.harn_schema_name
1170 }
1171
1172 fn metadata(&self) -> ProviderMetadata {
1173 self.metadata.clone()
1174 }
1175
1176 fn normalize(
1177 &self,
1178 kind: &str,
1179 headers: &BTreeMap<String, String>,
1180 raw: JsonValue,
1181 ) -> Result<ProviderPayload, ProviderCatalogError> {
1182 Ok((self.normalize)(kind, headers, raw))
1183 }
1184}
1185
1186fn provider_metadata_entry(
1187 provider: &str,
1188 kinds: &[&str],
1189 schema_name: &str,
1190 outbound_methods: &[&str],
1191 signature_verification: SignatureVerificationMetadata,
1192 secret_requirements: Vec<ProviderSecretRequirement>,
1193 runtime: ProviderRuntimeMetadata,
1194) -> ProviderMetadata {
1195 ProviderMetadata {
1196 provider: provider.to_string(),
1197 kinds: kinds.iter().map(|kind| kind.to_string()).collect(),
1198 schema_name: schema_name.to_string(),
1199 outbound_methods: outbound_methods
1200 .iter()
1201 .map(|name| ProviderOutboundMethod {
1202 name: (*name).to_string(),
1203 })
1204 .collect(),
1205 secret_requirements,
1206 signature_verification,
1207 runtime,
1208 }
1209}
1210
1211fn hmac_signature_metadata(
1212 variant: &str,
1213 signature_header: &str,
1214 timestamp_header: Option<&str>,
1215 id_header: Option<&str>,
1216 default_tolerance_secs: Option<i64>,
1217 encoding: &str,
1218) -> SignatureVerificationMetadata {
1219 SignatureVerificationMetadata::Hmac {
1220 variant: variant.to_string(),
1221 raw_body: true,
1222 signature_header: signature_header.to_string(),
1223 timestamp_header: timestamp_header.map(ToString::to_string),
1224 id_header: id_header.map(ToString::to_string),
1225 default_tolerance_secs,
1226 digest: "sha256".to_string(),
1227 encoding: encoding.to_string(),
1228 }
1229}
1230
1231fn required_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
1232 ProviderSecretRequirement {
1233 name: name.to_string(),
1234 required: true,
1235 namespace: namespace.to_string(),
1236 }
1237}
1238
1239fn outbound_method(name: &str) -> ProviderOutboundMethod {
1240 ProviderOutboundMethod {
1241 name: name.to_string(),
1242 }
1243}
1244
1245fn optional_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
1246 ProviderSecretRequirement {
1247 name: name.to_string(),
1248 required: false,
1249 namespace: namespace.to_string(),
1250 }
1251}
1252
1253fn default_provider_schemas() -> Vec<Arc<dyn ProviderSchema>> {
1254 vec![
1255 Arc::new(BuiltinProviderSchema {
1256 provider_id: "github",
1257 harn_schema_name: "GitHubEventPayload",
1258 metadata: provider_metadata_entry(
1259 "github",
1260 &["webhook"],
1261 "GitHubEventPayload",
1262 &[],
1263 hmac_signature_metadata(
1264 "github",
1265 "X-Hub-Signature-256",
1266 None,
1267 Some("X-GitHub-Delivery"),
1268 None,
1269 "hex",
1270 ),
1271 vec![required_secret("signing_secret", "github")],
1272 ProviderRuntimeMetadata::Placeholder,
1273 ),
1274 normalize: github_payload,
1275 }),
1276 Arc::new(BuiltinProviderSchema {
1277 provider_id: "slack",
1278 harn_schema_name: "SlackEventPayload",
1279 metadata: provider_metadata_entry(
1280 "slack",
1281 &["webhook"],
1282 "SlackEventPayload",
1283 &[
1284 "post_message",
1285 "update_message",
1286 "add_reaction",
1287 "open_view",
1288 "user_info",
1289 "api_call",
1290 "upload_file",
1291 ],
1292 hmac_signature_metadata(
1293 "slack",
1294 "X-Slack-Signature",
1295 Some("X-Slack-Request-Timestamp"),
1296 None,
1297 Some(300),
1298 "hex",
1299 ),
1300 vec![required_secret("signing_secret", "slack")],
1301 ProviderRuntimeMetadata::Placeholder,
1302 ),
1303 normalize: slack_payload,
1304 }),
1305 Arc::new(BuiltinProviderSchema {
1306 provider_id: "linear",
1307 harn_schema_name: "LinearEventPayload",
1308 metadata: {
1309 let mut metadata = provider_metadata_entry(
1310 "linear",
1311 &["webhook"],
1312 "LinearEventPayload",
1313 &[],
1314 hmac_signature_metadata(
1315 "linear",
1316 "Linear-Signature",
1317 None,
1318 Some("Linear-Delivery"),
1319 Some(75),
1320 "hex",
1321 ),
1322 vec![
1323 required_secret("signing_secret", "linear"),
1324 optional_secret("access_token", "linear"),
1325 ],
1326 ProviderRuntimeMetadata::Placeholder,
1327 );
1328 metadata.outbound_methods = vec![
1329 ProviderOutboundMethod {
1330 name: "list_issues".to_string(),
1331 },
1332 ProviderOutboundMethod {
1333 name: "update_issue".to_string(),
1334 },
1335 ProviderOutboundMethod {
1336 name: "create_comment".to_string(),
1337 },
1338 ProviderOutboundMethod {
1339 name: "search".to_string(),
1340 },
1341 ProviderOutboundMethod {
1342 name: "graphql".to_string(),
1343 },
1344 ];
1345 metadata
1346 },
1347 normalize: linear_payload,
1348 }),
1349 Arc::new(BuiltinProviderSchema {
1350 provider_id: "notion",
1351 harn_schema_name: "NotionEventPayload",
1352 metadata: {
1353 let mut metadata = provider_metadata_entry(
1354 "notion",
1355 &["webhook", "poll"],
1356 "NotionEventPayload",
1357 &[],
1358 hmac_signature_metadata(
1359 "notion",
1360 "X-Notion-Signature",
1361 None,
1362 None,
1363 None,
1364 "hex",
1365 ),
1366 vec![required_secret("verification_token", "notion")],
1367 ProviderRuntimeMetadata::Placeholder,
1368 );
1369 metadata.outbound_methods = vec![
1370 outbound_method("get_page"),
1371 outbound_method("update_page"),
1372 outbound_method("append_blocks"),
1373 outbound_method("query_database"),
1374 outbound_method("search"),
1375 outbound_method("create_comment"),
1376 outbound_method("api_call"),
1377 ];
1378 metadata
1379 },
1380 normalize: notion_payload,
1381 }),
1382 Arc::new(BuiltinProviderSchema {
1383 provider_id: "cron",
1384 harn_schema_name: "CronEventPayload",
1385 metadata: provider_metadata_entry(
1386 "cron",
1387 &["cron"],
1388 "CronEventPayload",
1389 &[],
1390 SignatureVerificationMetadata::None,
1391 Vec::new(),
1392 ProviderRuntimeMetadata::Builtin {
1393 connector: "cron".to_string(),
1394 default_signature_variant: None,
1395 },
1396 ),
1397 normalize: cron_payload,
1398 }),
1399 Arc::new(BuiltinProviderSchema {
1400 provider_id: "webhook",
1401 harn_schema_name: "GenericWebhookPayload",
1402 metadata: provider_metadata_entry(
1403 "webhook",
1404 &["webhook"],
1405 "GenericWebhookPayload",
1406 &[],
1407 hmac_signature_metadata(
1408 "standard",
1409 "webhook-signature",
1410 Some("webhook-timestamp"),
1411 Some("webhook-id"),
1412 Some(300),
1413 "base64",
1414 ),
1415 vec![required_secret("signing_secret", "webhook")],
1416 ProviderRuntimeMetadata::Builtin {
1417 connector: "webhook".to_string(),
1418 default_signature_variant: Some("standard".to_string()),
1419 },
1420 ),
1421 normalize: webhook_payload,
1422 }),
1423 Arc::new(BuiltinProviderSchema {
1424 provider_id: "a2a-push",
1425 harn_schema_name: "A2aPushPayload",
1426 metadata: provider_metadata_entry(
1427 "a2a-push",
1428 &["a2a-push"],
1429 "A2aPushPayload",
1430 &[],
1431 SignatureVerificationMetadata::None,
1432 Vec::new(),
1433 ProviderRuntimeMetadata::Builtin {
1434 connector: "a2a-push".to_string(),
1435 default_signature_variant: None,
1436 },
1437 ),
1438 normalize: a2a_push_payload,
1439 }),
1440 Arc::new(stream_provider_schema("kafka", kafka_payload)),
1441 Arc::new(stream_provider_schema("nats", nats_payload)),
1442 Arc::new(stream_provider_schema("pulsar", pulsar_payload)),
1443 Arc::new(stream_provider_schema("postgres-cdc", postgres_cdc_payload)),
1444 Arc::new(stream_provider_schema("email", email_payload)),
1445 Arc::new(stream_provider_schema("websocket", websocket_payload)),
1446 ]
1447}
1448
1449fn stream_provider_schema(
1450 provider_id: &'static str,
1451 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
1452) -> BuiltinProviderSchema {
1453 BuiltinProviderSchema {
1454 provider_id,
1455 harn_schema_name: "StreamEventPayload",
1456 metadata: provider_metadata_entry(
1457 provider_id,
1458 &["stream"],
1459 "StreamEventPayload",
1460 &[],
1461 SignatureVerificationMetadata::None,
1462 Vec::new(),
1463 ProviderRuntimeMetadata::Builtin {
1464 connector: "stream".to_string(),
1465 default_signature_variant: None,
1466 },
1467 ),
1468 normalize,
1469 }
1470}
1471
1472fn github_payload(
1473 kind: &str,
1474 headers: &BTreeMap<String, String>,
1475 raw: JsonValue,
1476) -> ProviderPayload {
1477 let original_raw = raw
1482 .get("raw")
1483 .filter(|value| value.is_object())
1484 .cloned()
1485 .unwrap_or_else(|| raw.clone());
1486 let common = GitHubEventCommon {
1487 event: kind.to_string(),
1488 action: raw
1489 .get("action")
1490 .and_then(JsonValue::as_str)
1491 .map(ToString::to_string),
1492 delivery_id: raw
1493 .get("delivery_id")
1494 .and_then(JsonValue::as_str)
1495 .map(ToString::to_string)
1496 .or_else(|| headers.get("X-GitHub-Delivery").cloned()),
1497 installation_id: raw
1498 .get("installation_id")
1499 .and_then(JsonValue::as_i64)
1500 .or_else(|| {
1501 raw.get("installation")
1502 .and_then(|value| value.get("id"))
1503 .and_then(JsonValue::as_i64)
1504 }),
1505 topic: raw
1506 .get("topic")
1507 .and_then(JsonValue::as_str)
1508 .map(ToString::to_string),
1509 repository: raw.get("repository").cloned(),
1510 repo: raw.get("repo").cloned(),
1511 raw: original_raw,
1512 };
1513 let payload = match kind {
1514 "issues" => GitHubEventPayload::Issues(GitHubIssuesEventPayload {
1515 common,
1516 issue: raw.get("issue").cloned().unwrap_or(JsonValue::Null),
1517 }),
1518 "pull_request" => GitHubEventPayload::PullRequest(GitHubPullRequestEventPayload {
1519 common,
1520 pull_request: raw.get("pull_request").cloned().unwrap_or(JsonValue::Null),
1521 }),
1522 "issue_comment" => GitHubEventPayload::IssueComment(GitHubIssueCommentEventPayload {
1523 common,
1524 issue: raw.get("issue").cloned().unwrap_or(JsonValue::Null),
1525 comment: raw.get("comment").cloned().unwrap_or(JsonValue::Null),
1526 }),
1527 "pull_request_review" => {
1528 GitHubEventPayload::PullRequestReview(GitHubPullRequestReviewEventPayload {
1529 common,
1530 pull_request: raw.get("pull_request").cloned().unwrap_or(JsonValue::Null),
1531 review: raw.get("review").cloned().unwrap_or(JsonValue::Null),
1532 })
1533 }
1534 "push" => GitHubEventPayload::Push(GitHubPushEventPayload {
1535 common,
1536 commits: raw
1537 .get("commits")
1538 .and_then(JsonValue::as_array)
1539 .cloned()
1540 .unwrap_or_default(),
1541 distinct_size: raw.get("distinct_size").and_then(JsonValue::as_i64),
1542 }),
1543 "workflow_run" => GitHubEventPayload::WorkflowRun(GitHubWorkflowRunEventPayload {
1544 common,
1545 workflow_run: raw.get("workflow_run").cloned().unwrap_or(JsonValue::Null),
1546 }),
1547 "deployment_status" => {
1548 GitHubEventPayload::DeploymentStatus(GitHubDeploymentStatusEventPayload {
1549 common,
1550 deployment_status: raw
1551 .get("deployment_status")
1552 .cloned()
1553 .unwrap_or(JsonValue::Null),
1554 deployment: raw.get("deployment").cloned().unwrap_or(JsonValue::Null),
1555 })
1556 }
1557 "check_run" => GitHubEventPayload::CheckRun(GitHubCheckRunEventPayload {
1558 common,
1559 check_run: raw.get("check_run").cloned().unwrap_or(JsonValue::Null),
1560 }),
1561 "check_suite" => {
1562 let check_suite = raw.get("check_suite").cloned().unwrap_or(JsonValue::Null);
1563 GitHubEventPayload::CheckSuite(GitHubCheckSuiteEventPayload {
1564 check_suite_id: github_promoted_i64(&raw, "check_suite_id")
1565 .or_else(|| check_suite.get("id").and_then(JsonValue::as_i64)),
1566 pull_request_number: github_promoted_i64(&raw, "pull_request_number"),
1567 head_sha: github_promoted_string(&raw, "head_sha"),
1568 head_ref: github_promoted_string(&raw, "head_ref"),
1569 base_ref: github_promoted_string(&raw, "base_ref"),
1570 status: github_promoted_string(&raw, "status"),
1571 conclusion: github_promoted_string(&raw, "conclusion"),
1572 common,
1573 check_suite,
1574 })
1575 }
1576 "status" => GitHubEventPayload::Status(GitHubStatusEventPayload {
1577 commit_status: raw
1578 .get("commit_status")
1579 .cloned()
1580 .or_else(|| Some(common.raw.clone())),
1581 status_id: github_promoted_i64(&raw, "status_id")
1582 .or_else(|| common.raw.get("id").and_then(JsonValue::as_i64)),
1583 head_sha: github_promoted_string(&raw, "head_sha").or_else(|| {
1584 common
1585 .raw
1586 .get("sha")
1587 .and_then(JsonValue::as_str)
1588 .map(ToString::to_string)
1589 }),
1590 head_ref: github_promoted_string(&raw, "head_ref"),
1591 base_ref: github_promoted_string(&raw, "base_ref"),
1592 state: github_promoted_string(&raw, "state"),
1593 context: github_promoted_string(&raw, "context"),
1594 target_url: github_promoted_string(&raw, "target_url"),
1595 common,
1596 }),
1597 "merge_group" => {
1598 let merge_group = raw.get("merge_group").cloned().unwrap_or(JsonValue::Null);
1599 GitHubEventPayload::MergeGroup(GitHubMergeGroupEventPayload {
1600 merge_group_id: raw
1601 .get("merge_group_id")
1602 .cloned()
1603 .or_else(|| merge_group.get("id").cloned()),
1604 head_sha: github_promoted_string(&raw, "head_sha").or_else(|| {
1605 merge_group
1606 .get("head_sha")
1607 .and_then(JsonValue::as_str)
1608 .map(ToString::to_string)
1609 }),
1610 head_ref: github_promoted_string(&raw, "head_ref").or_else(|| {
1611 merge_group
1612 .get("head_ref")
1613 .and_then(JsonValue::as_str)
1614 .map(ToString::to_string)
1615 }),
1616 base_sha: github_promoted_string(&raw, "base_sha").or_else(|| {
1617 merge_group
1618 .get("base_sha")
1619 .and_then(JsonValue::as_str)
1620 .map(ToString::to_string)
1621 }),
1622 base_ref: github_promoted_string(&raw, "base_ref").or_else(|| {
1623 merge_group
1624 .get("base_ref")
1625 .and_then(JsonValue::as_str)
1626 .map(ToString::to_string)
1627 }),
1628 pull_requests: raw
1629 .get("pull_requests")
1630 .and_then(JsonValue::as_array)
1631 .cloned()
1632 .unwrap_or_default(),
1633 pull_request_numbers: raw
1634 .get("pull_request_numbers")
1635 .and_then(JsonValue::as_array)
1636 .map(|values| {
1637 values
1638 .iter()
1639 .filter_map(JsonValue::as_i64)
1640 .collect::<Vec<_>>()
1641 })
1642 .unwrap_or_default(),
1643 common,
1644 merge_group,
1645 })
1646 }
1647 "installation" => GitHubEventPayload::Installation(GitHubInstallationEventPayload {
1648 installation: raw.get("installation").cloned(),
1649 account: raw.get("account").cloned(),
1650 installation_state: github_promoted_string(&raw, "installation_state"),
1651 suspended: raw.get("suspended").and_then(JsonValue::as_bool),
1652 revoked: raw.get("revoked").and_then(JsonValue::as_bool),
1653 repositories: raw
1654 .get("repositories")
1655 .and_then(JsonValue::as_array)
1656 .cloned()
1657 .unwrap_or_default(),
1658 common,
1659 }),
1660 "installation_repositories" => GitHubEventPayload::InstallationRepositories(
1661 GitHubInstallationRepositoriesEventPayload {
1662 installation: raw.get("installation").cloned(),
1663 account: raw.get("account").cloned(),
1664 installation_state: github_promoted_string(&raw, "installation_state"),
1665 suspended: raw.get("suspended").and_then(JsonValue::as_bool),
1666 revoked: raw.get("revoked").and_then(JsonValue::as_bool),
1667 repository_selection: github_promoted_string(&raw, "repository_selection"),
1668 repositories_added: raw
1669 .get("repositories_added")
1670 .and_then(JsonValue::as_array)
1671 .cloned()
1672 .unwrap_or_default(),
1673 repositories_removed: raw
1674 .get("repositories_removed")
1675 .and_then(JsonValue::as_array)
1676 .cloned()
1677 .unwrap_or_default(),
1678 common,
1679 },
1680 ),
1681 _ => GitHubEventPayload::Other(common),
1682 };
1683 ProviderPayload::Known(KnownProviderPayload::GitHub(payload))
1684}
1685
1686fn github_promoted_string(raw: &JsonValue, field: &str) -> Option<String> {
1687 raw.get(field)
1688 .and_then(JsonValue::as_str)
1689 .map(ToString::to_string)
1690}
1691
1692fn github_promoted_i64(raw: &JsonValue, field: &str) -> Option<i64> {
1693 raw.get(field).and_then(JsonValue::as_i64)
1694}
1695
1696fn slack_payload(
1697 kind: &str,
1698 _headers: &BTreeMap<String, String>,
1699 raw: JsonValue,
1700) -> ProviderPayload {
1701 let event = raw.get("event");
1702 let common = SlackEventCommon {
1703 event: kind.to_string(),
1704 event_id: raw
1705 .get("event_id")
1706 .and_then(JsonValue::as_str)
1707 .map(ToString::to_string),
1708 api_app_id: raw
1709 .get("api_app_id")
1710 .and_then(JsonValue::as_str)
1711 .map(ToString::to_string),
1712 team_id: raw
1713 .get("team_id")
1714 .and_then(JsonValue::as_str)
1715 .map(ToString::to_string),
1716 channel_id: slack_channel_id(event),
1717 user_id: slack_user_id(event),
1718 event_ts: event
1719 .and_then(|value| value.get("event_ts"))
1720 .and_then(JsonValue::as_str)
1721 .map(ToString::to_string),
1722 raw: raw.clone(),
1723 };
1724 let payload = match kind {
1725 kind if kind == "message" || kind.starts_with("message.") => {
1726 SlackEventPayload::Message(SlackMessageEventPayload {
1727 subtype: event
1728 .and_then(|value| value.get("subtype"))
1729 .and_then(JsonValue::as_str)
1730 .map(ToString::to_string),
1731 channel_type: event
1732 .and_then(|value| value.get("channel_type"))
1733 .and_then(JsonValue::as_str)
1734 .map(ToString::to_string),
1735 channel: event
1736 .and_then(|value| value.get("channel"))
1737 .and_then(JsonValue::as_str)
1738 .map(ToString::to_string),
1739 user: event
1740 .and_then(|value| value.get("user"))
1741 .and_then(JsonValue::as_str)
1742 .map(ToString::to_string),
1743 text: event
1744 .and_then(|value| value.get("text"))
1745 .and_then(JsonValue::as_str)
1746 .map(ToString::to_string),
1747 ts: event
1748 .and_then(|value| value.get("ts"))
1749 .and_then(JsonValue::as_str)
1750 .map(ToString::to_string),
1751 thread_ts: event
1752 .and_then(|value| value.get("thread_ts"))
1753 .and_then(JsonValue::as_str)
1754 .map(ToString::to_string),
1755 common,
1756 })
1757 }
1758 "app_mention" => SlackEventPayload::AppMention(SlackAppMentionEventPayload {
1759 channel: event
1760 .and_then(|value| value.get("channel"))
1761 .and_then(JsonValue::as_str)
1762 .map(ToString::to_string),
1763 user: event
1764 .and_then(|value| value.get("user"))
1765 .and_then(JsonValue::as_str)
1766 .map(ToString::to_string),
1767 text: event
1768 .and_then(|value| value.get("text"))
1769 .and_then(JsonValue::as_str)
1770 .map(ToString::to_string),
1771 ts: event
1772 .and_then(|value| value.get("ts"))
1773 .and_then(JsonValue::as_str)
1774 .map(ToString::to_string),
1775 thread_ts: event
1776 .and_then(|value| value.get("thread_ts"))
1777 .and_then(JsonValue::as_str)
1778 .map(ToString::to_string),
1779 common,
1780 }),
1781 "reaction_added" => SlackEventPayload::ReactionAdded(SlackReactionAddedEventPayload {
1782 reaction: event
1783 .and_then(|value| value.get("reaction"))
1784 .and_then(JsonValue::as_str)
1785 .map(ToString::to_string),
1786 item_user: event
1787 .and_then(|value| value.get("item_user"))
1788 .and_then(JsonValue::as_str)
1789 .map(ToString::to_string),
1790 item: event
1791 .and_then(|value| value.get("item"))
1792 .cloned()
1793 .unwrap_or(JsonValue::Null),
1794 common,
1795 }),
1796 "app_home_opened" => SlackEventPayload::AppHomeOpened(SlackAppHomeOpenedEventPayload {
1797 user: event
1798 .and_then(|value| value.get("user"))
1799 .and_then(JsonValue::as_str)
1800 .map(ToString::to_string),
1801 channel: event
1802 .and_then(|value| value.get("channel"))
1803 .and_then(JsonValue::as_str)
1804 .map(ToString::to_string),
1805 tab: event
1806 .and_then(|value| value.get("tab"))
1807 .and_then(JsonValue::as_str)
1808 .map(ToString::to_string),
1809 view: event
1810 .and_then(|value| value.get("view"))
1811 .cloned()
1812 .unwrap_or(JsonValue::Null),
1813 common,
1814 }),
1815 "assistant_thread_started" => {
1816 let assistant_thread = event
1817 .and_then(|value| value.get("assistant_thread"))
1818 .cloned()
1819 .unwrap_or(JsonValue::Null);
1820 SlackEventPayload::AssistantThreadStarted(SlackAssistantThreadStartedEventPayload {
1821 thread_ts: assistant_thread
1822 .get("thread_ts")
1823 .and_then(JsonValue::as_str)
1824 .map(ToString::to_string),
1825 context: assistant_thread
1826 .get("context")
1827 .cloned()
1828 .unwrap_or(JsonValue::Null),
1829 assistant_thread,
1830 common,
1831 })
1832 }
1833 _ => SlackEventPayload::Other(common),
1834 };
1835 ProviderPayload::Known(KnownProviderPayload::Slack(Box::new(payload)))
1836}
1837
1838fn slack_channel_id(event: Option<&JsonValue>) -> Option<String> {
1839 event
1840 .and_then(|value| value.get("channel"))
1841 .and_then(JsonValue::as_str)
1842 .map(ToString::to_string)
1843 .or_else(|| {
1844 event
1845 .and_then(|value| value.get("item"))
1846 .and_then(|value| value.get("channel"))
1847 .and_then(JsonValue::as_str)
1848 .map(ToString::to_string)
1849 })
1850 .or_else(|| {
1851 event
1852 .and_then(|value| value.get("channel"))
1853 .and_then(|value| value.get("id"))
1854 .and_then(JsonValue::as_str)
1855 .map(ToString::to_string)
1856 })
1857 .or_else(|| {
1858 event
1859 .and_then(|value| value.get("assistant_thread"))
1860 .and_then(|value| value.get("channel_id"))
1861 .and_then(JsonValue::as_str)
1862 .map(ToString::to_string)
1863 })
1864}
1865
1866fn slack_user_id(event: Option<&JsonValue>) -> Option<String> {
1867 event
1868 .and_then(|value| value.get("user"))
1869 .and_then(JsonValue::as_str)
1870 .map(ToString::to_string)
1871 .or_else(|| {
1872 event
1873 .and_then(|value| value.get("user"))
1874 .and_then(|value| value.get("id"))
1875 .and_then(JsonValue::as_str)
1876 .map(ToString::to_string)
1877 })
1878 .or_else(|| {
1879 event
1880 .and_then(|value| value.get("item_user"))
1881 .and_then(JsonValue::as_str)
1882 .map(ToString::to_string)
1883 })
1884 .or_else(|| {
1885 event
1886 .and_then(|value| value.get("assistant_thread"))
1887 .and_then(|value| value.get("user_id"))
1888 .and_then(JsonValue::as_str)
1889 .map(ToString::to_string)
1890 })
1891}
1892
1893fn linear_payload(
1894 _kind: &str,
1895 headers: &BTreeMap<String, String>,
1896 raw: JsonValue,
1897) -> ProviderPayload {
1898 let common = linear_event_common(headers, &raw);
1899 let event = common.event.clone();
1900 let payload = match event.as_str() {
1901 "issue" => LinearEventPayload::Issue(LinearIssueEventPayload {
1902 common,
1903 issue: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1904 changes: parse_linear_issue_changes(raw.get("updatedFrom")),
1905 }),
1906 "comment" => LinearEventPayload::IssueComment(LinearIssueCommentEventPayload {
1907 common,
1908 comment: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1909 }),
1910 "issue_label" => LinearEventPayload::IssueLabel(LinearIssueLabelEventPayload {
1911 common,
1912 label: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1913 }),
1914 "project" => LinearEventPayload::Project(LinearProjectEventPayload {
1915 common,
1916 project: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1917 }),
1918 "cycle" => LinearEventPayload::Cycle(LinearCycleEventPayload {
1919 common,
1920 cycle: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1921 }),
1922 "customer" => LinearEventPayload::Customer(LinearCustomerEventPayload {
1923 common,
1924 customer: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1925 }),
1926 "customer_request" => {
1927 LinearEventPayload::CustomerRequest(LinearCustomerRequestEventPayload {
1928 common,
1929 customer_request: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1930 })
1931 }
1932 _ => LinearEventPayload::Other(common),
1933 };
1934 ProviderPayload::Known(KnownProviderPayload::Linear(payload))
1935}
1936
1937fn linear_event_common(headers: &BTreeMap<String, String>, raw: &JsonValue) -> LinearEventCommon {
1938 LinearEventCommon {
1939 event: linear_event_name(
1940 raw.get("type")
1941 .and_then(JsonValue::as_str)
1942 .or_else(|| headers.get("Linear-Event").map(String::as_str)),
1943 ),
1944 action: raw
1945 .get("action")
1946 .and_then(JsonValue::as_str)
1947 .map(ToString::to_string),
1948 delivery_id: header_value(headers, "Linear-Delivery").map(ToString::to_string),
1949 organization_id: raw
1950 .get("organizationId")
1951 .and_then(JsonValue::as_str)
1952 .map(ToString::to_string),
1953 webhook_timestamp: raw.get("webhookTimestamp").and_then(parse_json_i64ish),
1954 webhook_id: raw
1955 .get("webhookId")
1956 .and_then(JsonValue::as_str)
1957 .map(ToString::to_string),
1958 url: raw
1959 .get("url")
1960 .and_then(JsonValue::as_str)
1961 .map(ToString::to_string),
1962 created_at: raw
1963 .get("createdAt")
1964 .and_then(JsonValue::as_str)
1965 .map(ToString::to_string),
1966 actor: raw.get("actor").cloned().unwrap_or(JsonValue::Null),
1967 raw: raw.clone(),
1968 }
1969}
1970
1971fn linear_event_name(raw_type: Option<&str>) -> String {
1972 match raw_type.unwrap_or_default().to_ascii_lowercase().as_str() {
1973 "issue" => "issue".to_string(),
1974 "comment" | "issuecomment" | "issue_comment" => "comment".to_string(),
1975 "issuelabel" | "issue_label" => "issue_label".to_string(),
1976 "project" | "projectupdate" | "project_update" => "project".to_string(),
1977 "cycle" => "cycle".to_string(),
1978 "customer" => "customer".to_string(),
1979 "customerrequest" | "customer_request" => "customer_request".to_string(),
1980 other if !other.is_empty() => other.to_string(),
1981 _ => "other".to_string(),
1982 }
1983}
1984
1985fn parse_linear_issue_changes(updated_from: Option<&JsonValue>) -> Vec<LinearIssueChange> {
1986 let Some(JsonValue::Object(fields)) = updated_from else {
1987 return Vec::new();
1988 };
1989 let mut changes = Vec::new();
1990 for (field, previous) in fields {
1991 let change = match field.as_str() {
1992 "title" => LinearIssueChange::Title {
1993 previous: previous.as_str().map(ToString::to_string),
1994 },
1995 "description" => LinearIssueChange::Description {
1996 previous: previous.as_str().map(ToString::to_string),
1997 },
1998 "priority" => LinearIssueChange::Priority {
1999 previous: parse_json_i64ish(previous),
2000 },
2001 "estimate" => LinearIssueChange::Estimate {
2002 previous: parse_json_i64ish(previous),
2003 },
2004 "stateId" => LinearIssueChange::StateId {
2005 previous: previous.as_str().map(ToString::to_string),
2006 },
2007 "teamId" => LinearIssueChange::TeamId {
2008 previous: previous.as_str().map(ToString::to_string),
2009 },
2010 "assigneeId" => LinearIssueChange::AssigneeId {
2011 previous: previous.as_str().map(ToString::to_string),
2012 },
2013 "projectId" => LinearIssueChange::ProjectId {
2014 previous: previous.as_str().map(ToString::to_string),
2015 },
2016 "cycleId" => LinearIssueChange::CycleId {
2017 previous: previous.as_str().map(ToString::to_string),
2018 },
2019 "dueDate" => LinearIssueChange::DueDate {
2020 previous: previous.as_str().map(ToString::to_string),
2021 },
2022 "parentId" => LinearIssueChange::ParentId {
2023 previous: previous.as_str().map(ToString::to_string),
2024 },
2025 "sortOrder" => LinearIssueChange::SortOrder {
2026 previous: previous.as_f64(),
2027 },
2028 "labelIds" => LinearIssueChange::LabelIds {
2029 previous: parse_string_array(previous),
2030 },
2031 "completedAt" => LinearIssueChange::CompletedAt {
2032 previous: previous.as_str().map(ToString::to_string),
2033 },
2034 _ => LinearIssueChange::Other {
2035 field: field.clone(),
2036 previous: previous.clone(),
2037 },
2038 };
2039 changes.push(change);
2040 }
2041 changes
2042}
2043
2044fn parse_json_i64ish(value: &JsonValue) -> Option<i64> {
2045 value
2046 .as_i64()
2047 .or_else(|| value.as_u64().and_then(|raw| i64::try_from(raw).ok()))
2048 .or_else(|| value.as_str().and_then(|raw| raw.parse::<i64>().ok()))
2049}
2050
2051fn parse_string_array(value: &JsonValue) -> Vec<String> {
2052 let Some(array) = value.as_array() else {
2053 return Vec::new();
2054 };
2055 array
2056 .iter()
2057 .filter_map(|entry| {
2058 entry.as_str().map(ToString::to_string).or_else(|| {
2059 entry
2060 .get("id")
2061 .and_then(JsonValue::as_str)
2062 .map(ToString::to_string)
2063 })
2064 })
2065 .collect()
2066}
2067
2068fn header_value<'a>(headers: &'a BTreeMap<String, String>, name: &str) -> Option<&'a str> {
2069 headers
2070 .iter()
2071 .find(|(key, _)| key.eq_ignore_ascii_case(name))
2072 .map(|(_, value)| value.as_str())
2073}
2074
2075fn notion_payload(
2076 kind: &str,
2077 headers: &BTreeMap<String, String>,
2078 raw: JsonValue,
2079) -> ProviderPayload {
2080 let workspace_id = raw
2081 .get("workspace_id")
2082 .and_then(JsonValue::as_str)
2083 .map(ToString::to_string);
2084 ProviderPayload::Known(KnownProviderPayload::Notion(Box::new(NotionEventPayload {
2085 event: kind.to_string(),
2086 workspace_id,
2087 request_id: headers
2088 .get("request-id")
2089 .cloned()
2090 .or_else(|| headers.get("x-request-id").cloned()),
2091 subscription_id: raw
2092 .get("subscription_id")
2093 .and_then(JsonValue::as_str)
2094 .map(ToString::to_string),
2095 integration_id: raw
2096 .get("integration_id")
2097 .and_then(JsonValue::as_str)
2098 .map(ToString::to_string),
2099 attempt_number: raw
2100 .get("attempt_number")
2101 .and_then(JsonValue::as_u64)
2102 .and_then(|value| u32::try_from(value).ok()),
2103 entity_id: raw
2104 .get("entity")
2105 .and_then(|value| value.get("id"))
2106 .and_then(JsonValue::as_str)
2107 .map(ToString::to_string),
2108 entity_type: raw
2109 .get("entity")
2110 .and_then(|value| value.get("type"))
2111 .and_then(JsonValue::as_str)
2112 .map(ToString::to_string),
2113 api_version: raw
2114 .get("api_version")
2115 .and_then(JsonValue::as_str)
2116 .map(ToString::to_string),
2117 verification_token: raw
2118 .get("verification_token")
2119 .and_then(JsonValue::as_str)
2120 .map(ToString::to_string),
2121 polled: None,
2122 raw,
2123 })))
2124}
2125
2126fn cron_payload(
2127 _kind: &str,
2128 _headers: &BTreeMap<String, String>,
2129 raw: JsonValue,
2130) -> ProviderPayload {
2131 let cron_id = raw
2132 .get("cron_id")
2133 .and_then(JsonValue::as_str)
2134 .map(ToString::to_string);
2135 let schedule = raw
2136 .get("schedule")
2137 .and_then(JsonValue::as_str)
2138 .map(ToString::to_string);
2139 let tick_at = raw
2140 .get("tick_at")
2141 .and_then(JsonValue::as_str)
2142 .and_then(parse_rfc3339)
2143 .unwrap_or_else(OffsetDateTime::now_utc);
2144 ProviderPayload::Known(KnownProviderPayload::Cron(CronEventPayload {
2145 cron_id,
2146 schedule,
2147 tick_at,
2148 raw,
2149 }))
2150}
2151
2152fn webhook_payload(
2153 _kind: &str,
2154 headers: &BTreeMap<String, String>,
2155 raw: JsonValue,
2156) -> ProviderPayload {
2157 ProviderPayload::Known(KnownProviderPayload::Webhook(GenericWebhookPayload {
2158 source: headers.get("X-Webhook-Source").cloned(),
2159 content_type: headers.get("Content-Type").cloned(),
2160 raw,
2161 }))
2162}
2163
2164fn a2a_push_payload(
2165 _kind: &str,
2166 _headers: &BTreeMap<String, String>,
2167 raw: JsonValue,
2168) -> ProviderPayload {
2169 let task_id = raw
2170 .get("task_id")
2171 .and_then(JsonValue::as_str)
2172 .map(ToString::to_string);
2173 let sender = raw
2174 .get("sender")
2175 .and_then(JsonValue::as_str)
2176 .map(ToString::to_string);
2177 let task_state = raw
2178 .pointer("/status/state")
2179 .or_else(|| raw.pointer("/statusUpdate/status/state"))
2180 .and_then(JsonValue::as_str)
2181 .map(|state| match state {
2182 "cancelled" => "canceled".to_string(),
2183 other => other.to_string(),
2184 });
2185 let artifact = raw
2186 .pointer("/artifactUpdate/artifact")
2187 .or_else(|| raw.get("artifact"))
2188 .cloned();
2189 let kind = task_state
2190 .as_deref()
2191 .map(|state| format!("a2a.task.{state}"))
2192 .unwrap_or_else(|| "a2a.task.update".to_string());
2193 ProviderPayload::Known(KnownProviderPayload::A2aPush(A2aPushPayload {
2194 task_id,
2195 task_state,
2196 artifact,
2197 sender,
2198 raw,
2199 kind,
2200 }))
2201}
2202
2203fn kafka_payload(
2204 kind: &str,
2205 headers: &BTreeMap<String, String>,
2206 raw: JsonValue,
2207) -> ProviderPayload {
2208 ProviderPayload::Known(KnownProviderPayload::Kafka(stream_payload(
2209 kind, headers, raw,
2210 )))
2211}
2212
2213fn nats_payload(kind: &str, headers: &BTreeMap<String, String>, raw: JsonValue) -> ProviderPayload {
2214 ProviderPayload::Known(KnownProviderPayload::Nats(stream_payload(
2215 kind, headers, raw,
2216 )))
2217}
2218
2219fn pulsar_payload(
2220 kind: &str,
2221 headers: &BTreeMap<String, String>,
2222 raw: JsonValue,
2223) -> ProviderPayload {
2224 ProviderPayload::Known(KnownProviderPayload::Pulsar(stream_payload(
2225 kind, headers, raw,
2226 )))
2227}
2228
2229fn postgres_cdc_payload(
2230 kind: &str,
2231 headers: &BTreeMap<String, String>,
2232 raw: JsonValue,
2233) -> ProviderPayload {
2234 ProviderPayload::Known(KnownProviderPayload::PostgresCdc(stream_payload(
2235 kind, headers, raw,
2236 )))
2237}
2238
2239fn email_payload(
2240 kind: &str,
2241 headers: &BTreeMap<String, String>,
2242 raw: JsonValue,
2243) -> ProviderPayload {
2244 ProviderPayload::Known(KnownProviderPayload::Email(stream_payload(
2245 kind, headers, raw,
2246 )))
2247}
2248
2249fn websocket_payload(
2250 kind: &str,
2251 headers: &BTreeMap<String, String>,
2252 raw: JsonValue,
2253) -> ProviderPayload {
2254 ProviderPayload::Known(KnownProviderPayload::Websocket(stream_payload(
2255 kind, headers, raw,
2256 )))
2257}
2258
2259fn stream_payload(
2260 kind: &str,
2261 headers: &BTreeMap<String, String>,
2262 raw: JsonValue,
2263) -> StreamEventPayload {
2264 StreamEventPayload {
2265 event: kind.to_string(),
2266 source: json_stringish(&raw, &["source", "connector", "origin"]),
2267 stream: json_stringish(
2268 &raw,
2269 &["stream", "topic", "subject", "channel", "mailbox", "slot"],
2270 ),
2271 partition: json_stringish(&raw, &["partition", "shard", "consumer"]),
2272 offset: json_stringish(&raw, &["offset", "sequence", "lsn", "message_id"]),
2273 key: json_stringish(&raw, &["key", "message_key", "id", "event_id"]),
2274 timestamp: json_stringish(&raw, &["timestamp", "occurred_at", "received_at", "ts"]),
2275 headers: headers.clone(),
2276 raw,
2277 }
2278}
2279
2280fn json_stringish(raw: &JsonValue, fields: &[&str]) -> Option<String> {
2281 fields.iter().find_map(|field| {
2282 let value = raw.get(*field)?;
2283 value
2284 .as_str()
2285 .map(ToString::to_string)
2286 .or_else(|| parse_json_i64ish(value).map(|number| number.to_string()))
2287 .or_else(|| value.as_u64().map(|number| number.to_string()))
2288 })
2289}
2290
2291fn parse_rfc3339(text: &str) -> Option<OffsetDateTime> {
2292 OffsetDateTime::parse(text, &time::format_description::well_known::Rfc3339).ok()
2293}
2294
2295#[cfg(test)]
2296mod tests {
2297 use super::*;
2298
2299 struct OwnedProviderSchema {
2300 metadata: ProviderMetadata,
2301 }
2302
2303 impl OwnedProviderSchema {
2304 fn new(provider: &str, schema_name: &str) -> Self {
2305 Self {
2306 metadata: ProviderMetadata {
2307 provider: provider.to_string(),
2308 kinds: vec!["webhook".to_string()],
2309 schema_name: schema_name.to_string(),
2310 runtime: ProviderRuntimeMetadata::Placeholder,
2311 ..ProviderMetadata::default()
2312 },
2313 }
2314 }
2315 }
2316
2317 impl ProviderSchema for OwnedProviderSchema {
2318 fn provider_id(&self) -> &str {
2319 &self.metadata.provider
2320 }
2321
2322 fn harn_schema_name(&self) -> &str {
2323 &self.metadata.schema_name
2324 }
2325
2326 fn metadata(&self) -> ProviderMetadata {
2327 self.metadata.clone()
2328 }
2329
2330 fn normalize(
2331 &self,
2332 _kind: &str,
2333 _headers: &BTreeMap<String, String>,
2334 raw: JsonValue,
2335 ) -> Result<ProviderPayload, ProviderCatalogError> {
2336 Ok(ProviderPayload::Extension(ExtensionProviderPayload {
2337 provider: self.metadata.provider.clone(),
2338 schema_name: self.metadata.schema_name.clone(),
2339 raw,
2340 }))
2341 }
2342 }
2343
2344 fn owned_provider_schema(provider: &str, schema_name: &str) -> Arc<dyn ProviderSchema> {
2345 Arc::new(OwnedProviderSchema::new(provider, schema_name))
2346 }
2347
2348 fn sample_headers() -> BTreeMap<String, String> {
2349 BTreeMap::from([
2350 ("Authorization".to_string(), "Bearer secret".to_string()),
2351 ("Cookie".to_string(), "session=abc".to_string()),
2352 ("User-Agent".to_string(), "GitHub-Hookshot/123".to_string()),
2353 ("X-GitHub-Delivery".to_string(), "delivery-123".to_string()),
2354 ("X-GitHub-Event".to_string(), "issues".to_string()),
2355 ("X-Webhook-Token".to_string(), "token".to_string()),
2356 ])
2357 }
2358
2359 #[test]
2360 fn default_redaction_policy_keeps_safe_headers() {
2361 let redacted = redact_headers(&sample_headers(), &HeaderRedactionPolicy::default());
2362 assert_eq!(redacted.get("User-Agent").unwrap(), "GitHub-Hookshot/123");
2363 assert_eq!(redacted.get("X-GitHub-Delivery").unwrap(), "delivery-123");
2364 assert_eq!(
2365 redacted.get("Authorization").unwrap(),
2366 REDACTED_HEADER_VALUE
2367 );
2368 assert_eq!(redacted.get("Cookie").unwrap(), REDACTED_HEADER_VALUE);
2369 assert_eq!(
2370 redacted.get("X-Webhook-Token").unwrap(),
2371 REDACTED_HEADER_VALUE
2372 );
2373 }
2374
2375 #[test]
2376 fn provider_catalog_rejects_duplicates() {
2377 let mut catalog = ProviderCatalog::default();
2378 catalog
2379 .register(Arc::new(BuiltinProviderSchema {
2380 provider_id: "github",
2381 harn_schema_name: "GitHubEventPayload",
2382 metadata: provider_metadata_entry(
2383 "github",
2384 &["webhook"],
2385 "GitHubEventPayload",
2386 &[],
2387 SignatureVerificationMetadata::None,
2388 Vec::new(),
2389 ProviderRuntimeMetadata::Placeholder,
2390 ),
2391 normalize: github_payload,
2392 }))
2393 .unwrap();
2394 let error = catalog
2395 .register(Arc::new(BuiltinProviderSchema {
2396 provider_id: "github",
2397 harn_schema_name: "GitHubEventPayload",
2398 metadata: provider_metadata_entry(
2399 "github",
2400 &["webhook"],
2401 "GitHubEventPayload",
2402 &[],
2403 SignatureVerificationMetadata::None,
2404 Vec::new(),
2405 ProviderRuntimeMetadata::Placeholder,
2406 ),
2407 normalize: github_payload,
2408 }))
2409 .unwrap_err();
2410 assert_eq!(
2411 error,
2412 ProviderCatalogError::DuplicateProvider("github".to_string())
2413 );
2414 }
2415
2416 #[test]
2417 fn provider_catalog_builds_independent_owned_dynamic_catalogs() {
2418 let first = ProviderCatalog::with_defaults_and(vec![owned_provider_schema(
2419 "runtime-a",
2420 "RuntimeAPayload",
2421 )])
2422 .unwrap();
2423 assert_eq!(
2424 first
2425 .metadata_for("runtime-a")
2426 .expect("first dynamic provider")
2427 .schema_name,
2428 "RuntimeAPayload"
2429 );
2430 assert!(first.metadata_for("runtime-b").is_none());
2431
2432 let second = ProviderCatalog::with_defaults_and(vec![owned_provider_schema(
2433 "runtime-b",
2434 "RuntimeBPayload",
2435 )])
2436 .unwrap();
2437 assert!(second.metadata_for("runtime-a").is_none());
2438 assert_eq!(
2439 second
2440 .metadata_for("runtime-b")
2441 .expect("second dynamic provider")
2442 .schema_name,
2443 "RuntimeBPayload"
2444 );
2445 assert!(first.metadata_for("runtime-a").is_some());
2446 }
2447
2448 #[test]
2449 fn registered_provider_metadata_marks_builtin_connectors() {
2450 let entries = registered_provider_metadata();
2451 let builtin: Vec<&ProviderMetadata> = entries
2452 .iter()
2453 .filter(|entry| matches!(entry.runtime, ProviderRuntimeMetadata::Builtin { .. }))
2454 .collect();
2455
2456 assert_eq!(builtin.len(), 9);
2457 assert!(builtin.iter().any(|entry| entry.provider == "a2a-push"));
2458 assert!(builtin.iter().any(|entry| entry.provider == "cron"));
2459 assert!(builtin.iter().any(|entry| entry.provider == "webhook"));
2460 for provider in ["github", "linear", "notion", "slack"] {
2461 let entry = entries
2462 .iter()
2463 .find(|entry| entry.provider == provider)
2464 .expect("first-party package-backed provider metadata");
2465 assert!(matches!(
2466 entry.runtime,
2467 ProviderRuntimeMetadata::Placeholder
2468 ));
2469 }
2470 let kafka = entries
2471 .iter()
2472 .find(|entry| entry.provider == "kafka")
2473 .expect("kafka stream provider");
2474 assert_eq!(kafka.kinds, vec!["stream".to_string()]);
2475 assert_eq!(kafka.schema_name, "StreamEventPayload");
2476 assert!(matches!(
2477 kafka.runtime,
2478 ProviderRuntimeMetadata::Builtin {
2479 ref connector,
2480 default_signature_variant: None
2481 } if connector == "stream"
2482 ));
2483 }
2484
2485 #[test]
2486 fn trigger_event_round_trip_is_stable() {
2487 let provider = ProviderId::from("github");
2488 let headers = redact_headers(&sample_headers(), &HeaderRedactionPolicy::default());
2489 let payload = ProviderPayload::normalize(
2490 &provider,
2491 "issues",
2492 &sample_headers(),
2493 serde_json::json!({
2494 "action": "opened",
2495 "installation": {"id": 42},
2496 "issue": {"number": 99}
2497 }),
2498 )
2499 .unwrap();
2500 let event = TriggerEvent {
2501 id: TriggerEventId("trigger_evt_fixed".to_string()),
2502 provider,
2503 kind: "issues".to_string(),
2504 received_at: parse_rfc3339("2026-04-19T07:00:00Z").unwrap(),
2505 occurred_at: Some(parse_rfc3339("2026-04-19T06:59:59Z").unwrap()),
2506 dedupe_key: "delivery-123".to_string(),
2507 trace_id: TraceId("trace_fixed".to_string()),
2508 tenant_id: Some(TenantId("tenant_1".to_string())),
2509 headers,
2510 provider_payload: payload,
2511 signature_status: SignatureStatus::Verified,
2512 dedupe_claimed: false,
2513 batch: None,
2514 raw_body: Some(vec![0, 159, 255, 10]),
2515 };
2516
2517 let once = serde_json::to_value(&event).unwrap();
2518 assert_eq!(once["raw_body"], serde_json::json!("AJ//Cg=="));
2519 let decoded: TriggerEvent = serde_json::from_value(once.clone()).unwrap();
2520 let twice = serde_json::to_value(&decoded).unwrap();
2521 assert_eq!(decoded, event);
2522 assert_eq!(once, twice);
2523 }
2524
2525 #[test]
2526 fn unknown_provider_errors() {
2527 let error = ProviderPayload::normalize(
2528 &ProviderId::from("custom-provider"),
2529 "thing.happened",
2530 &BTreeMap::new(),
2531 serde_json::json!({"ok": true}),
2532 )
2533 .unwrap_err();
2534 assert_eq!(
2535 error,
2536 ProviderCatalogError::UnknownProvider("custom-provider".to_string())
2537 );
2538 }
2539
2540 fn github_headers(event: &str, delivery: &str) -> BTreeMap<String, String> {
2541 BTreeMap::from([
2542 ("X-GitHub-Event".to_string(), event.to_string()),
2543 ("X-GitHub-Delivery".to_string(), delivery.to_string()),
2544 ])
2545 }
2546
2547 fn unwrap_github(payload: ProviderPayload) -> GitHubEventPayload {
2548 match payload {
2549 ProviderPayload::Known(KnownProviderPayload::GitHub(p)) => p,
2550 other => panic!("expected GitHub payload, got {other:?}"),
2551 }
2552 }
2553
2554 fn connector_normalized(
2558 event: &str,
2559 delivery: &str,
2560 installation_id: i64,
2561 action: Option<&str>,
2562 original: serde_json::Value,
2563 promoted: serde_json::Value,
2564 ) -> serde_json::Value {
2565 let mut common = serde_json::json!({
2566 "provider": "github",
2567 "event": event,
2568 "topic": match action {
2569 Some(a) => format!("github.{event}.{a}"),
2570 None => format!("github.{event}"),
2571 },
2572 "delivery_id": delivery,
2573 "installation_id": installation_id,
2574 "repository": original.get("repository").cloned().unwrap_or(JsonValue::Null),
2575 "repo": serde_json::json!({"owner": "octo-org", "name": "octo-repo", "full_name": "octo-org/octo-repo"}),
2576 "raw": original,
2577 });
2578 if let Some(a) = action {
2579 common["action"] = serde_json::json!(a);
2580 }
2581 let common_obj = common.as_object_mut().unwrap();
2582 if let Some(promoted_obj) = promoted.as_object() {
2583 for (k, v) in promoted_obj {
2584 common_obj.insert(k.clone(), v.clone());
2585 }
2586 }
2587 common
2588 }
2589
2590 #[test]
2591 fn github_check_suite_event_promotes_typed_fields() {
2592 let original = serde_json::json!({
2593 "action": "requested",
2594 "check_suite": {
2595 "id": 8101,
2596 "status": "queued",
2597 "conclusion": null,
2598 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2599 "head_branch": "feature/x",
2600 },
2601 "repository": {"full_name": "octo-org/octo-repo"},
2602 "installation": {"id": 3001},
2603 });
2604 let normalized = connector_normalized(
2605 "check_suite",
2606 "delivery-cs",
2607 3001,
2608 Some("requested"),
2609 original.clone(),
2610 serde_json::json!({
2611 "check_suite": original["check_suite"].clone(),
2612 "check_suite_id": 8101,
2613 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2614 "head_ref": "feature/x",
2615 "status": "queued",
2616 }),
2617 );
2618 let provider = ProviderId::from("github");
2619 let payload = ProviderPayload::normalize(
2620 &provider,
2621 "check_suite",
2622 &github_headers("check_suite", "delivery-cs"),
2623 normalized,
2624 )
2625 .expect("check_suite payload");
2626 let GitHubEventPayload::CheckSuite(check_suite) = unwrap_github(payload) else {
2627 panic!("expected CheckSuite variant");
2628 };
2629 assert_eq!(check_suite.common.event, "check_suite");
2630 assert_eq!(check_suite.common.action.as_deref(), Some("requested"));
2631 assert_eq!(
2632 check_suite.common.delivery_id.as_deref(),
2633 Some("delivery-cs")
2634 );
2635 assert_eq!(check_suite.common.installation_id, Some(3001));
2636 assert_eq!(
2637 check_suite.common.topic.as_deref(),
2638 Some("github.check_suite.requested")
2639 );
2640 assert!(check_suite.common.repository.is_some());
2641 assert!(check_suite.common.repo.is_some());
2642 assert_eq!(check_suite.check_suite_id, Some(8101));
2643 assert_eq!(
2644 check_suite.head_sha.as_deref(),
2645 Some("ccccccccccccccccccccccccccccccccccccccc1")
2646 );
2647 assert_eq!(check_suite.head_ref.as_deref(), Some("feature/x"));
2648 assert_eq!(check_suite.status.as_deref(), Some("queued"));
2649 assert_eq!(check_suite.common.raw, original);
2651 }
2652
2653 #[test]
2654 fn github_status_event_promotes_typed_fields() {
2655 let original = serde_json::json!({
2656 "id": 9101,
2657 "sha": "ccccccccccccccccccccccccccccccccccccccc1",
2658 "state": "success",
2659 "context": "legacy/status",
2660 "target_url": "https://ci.example.test/octo-repo/9101",
2661 "branches": [{"name": "main"}],
2662 "repository": {"full_name": "octo-org/octo-repo"},
2663 "installation": {"id": 3001},
2664 });
2665 let normalized = connector_normalized(
2666 "status",
2667 "delivery-status",
2668 3001,
2669 None,
2670 original.clone(),
2671 serde_json::json!({
2672 "commit_status": original.clone(),
2673 "status_id": 9101,
2674 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2675 "head_ref": "main",
2676 "base_ref": "main",
2677 "state": "success",
2678 "context": "legacy/status",
2679 "target_url": "https://ci.example.test/octo-repo/9101",
2680 }),
2681 );
2682 let provider = ProviderId::from("github");
2683 let payload = ProviderPayload::normalize(
2684 &provider,
2685 "status",
2686 &github_headers("status", "delivery-status"),
2687 normalized,
2688 )
2689 .expect("status payload");
2690 let GitHubEventPayload::Status(status) = unwrap_github(payload) else {
2691 panic!("expected Status variant");
2692 };
2693 assert_eq!(status.common.event, "status");
2694 assert_eq!(status.common.installation_id, Some(3001));
2695 assert_eq!(status.status_id, Some(9101));
2696 assert_eq!(status.state.as_deref(), Some("success"));
2697 assert_eq!(status.context.as_deref(), Some("legacy/status"));
2698 assert_eq!(
2699 status.target_url.as_deref(),
2700 Some("https://ci.example.test/octo-repo/9101")
2701 );
2702 assert_eq!(
2703 status.head_sha.as_deref(),
2704 Some("ccccccccccccccccccccccccccccccccccccccc1")
2705 );
2706 assert!(status.commit_status.is_some());
2707 }
2708
2709 #[test]
2710 fn github_merge_group_event_promotes_typed_fields() {
2711 let original = serde_json::json!({
2712 "action": "checks_requested",
2713 "merge_group": {
2714 "id": 9201,
2715 "head_ref": "gh-readonly-queue/main/pr-42",
2716 "head_sha": "ddddddddddddddddddddddddddddddddddddddd1",
2717 "base_ref": "main",
2718 "base_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
2719 "pull_requests": [{"number": 42}, {"number": 43}],
2720 },
2721 "repository": {"full_name": "octo-org/octo-repo"},
2722 "installation": {"id": 3001},
2723 });
2724 let normalized = connector_normalized(
2725 "merge_group",
2726 "delivery-mg",
2727 3001,
2728 Some("checks_requested"),
2729 original.clone(),
2730 serde_json::json!({
2731 "merge_group": original["merge_group"].clone(),
2732 "merge_group_id": 9201,
2733 "head_sha": "ddddddddddddddddddddddddddddddddddddddd1",
2734 "head_ref": "gh-readonly-queue/main/pr-42",
2735 "base_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
2736 "base_ref": "main",
2737 "pull_requests": [{"number": 42}, {"number": 43}],
2738 "pull_request_numbers": [42, 43],
2739 }),
2740 );
2741 let provider = ProviderId::from("github");
2742 let payload = ProviderPayload::normalize(
2743 &provider,
2744 "merge_group",
2745 &github_headers("merge_group", "delivery-mg"),
2746 normalized,
2747 )
2748 .expect("merge_group payload");
2749 let GitHubEventPayload::MergeGroup(mg) = unwrap_github(payload) else {
2750 panic!("expected MergeGroup variant");
2751 };
2752 assert_eq!(mg.common.event, "merge_group");
2753 assert_eq!(mg.common.action.as_deref(), Some("checks_requested"));
2754 assert_eq!(mg.merge_group_id, Some(serde_json::json!(9201)));
2755 assert_eq!(mg.head_ref.as_deref(), Some("gh-readonly-queue/main/pr-42"));
2756 assert_eq!(
2757 mg.base_sha.as_deref(),
2758 Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1")
2759 );
2760 assert_eq!(mg.base_ref.as_deref(), Some("main"));
2761 assert_eq!(mg.pull_request_numbers, vec![42i64, 43i64]);
2762 assert_eq!(mg.pull_requests.len(), 2);
2763 }
2764
2765 #[test]
2766 fn github_installation_event_promotes_typed_fields() {
2767 let original = serde_json::json!({
2768 "action": "suspend",
2769 "installation": {
2770 "id": 3001,
2771 "account": {"login": "octo-org"},
2772 "repository_selection": "selected",
2773 "suspended_at": "2026-04-20T18:00:00Z",
2774 },
2775 "repositories": [{"full_name": "octo-org/octo-repo"}],
2776 });
2777 let normalized = connector_normalized(
2778 "installation",
2779 "delivery-inst",
2780 3001,
2781 Some("suspend"),
2782 original.clone(),
2783 serde_json::json!({
2784 "installation": original["installation"].clone(),
2785 "account": {"login": "octo-org"},
2786 "installation_state": "suspended",
2787 "suspended": true,
2788 "revoked": false,
2789 "repositories": original["repositories"].clone(),
2790 }),
2791 );
2792 let provider = ProviderId::from("github");
2793 let payload = ProviderPayload::normalize(
2794 &provider,
2795 "installation",
2796 &github_headers("installation", "delivery-inst"),
2797 normalized,
2798 )
2799 .expect("installation payload");
2800 let GitHubEventPayload::Installation(inst) = unwrap_github(payload) else {
2801 panic!("expected Installation variant");
2802 };
2803 assert_eq!(inst.common.event, "installation");
2804 assert_eq!(inst.common.action.as_deref(), Some("suspend"));
2805 assert_eq!(inst.installation_state.as_deref(), Some("suspended"));
2806 assert_eq!(inst.suspended, Some(true));
2807 assert_eq!(inst.revoked, Some(false));
2808 assert_eq!(inst.repositories.len(), 1);
2809 assert!(inst.account.is_some());
2810 }
2811
2812 #[test]
2813 fn github_installation_repositories_event_promotes_typed_fields() {
2814 let original = serde_json::json!({
2815 "action": "removed",
2816 "installation": {"id": 3001, "account": {"login": "octo-org"}},
2817 "repository_selection": "selected",
2818 "repositories_added": [],
2819 "repositories_removed": [
2820 {"id": 4001, "full_name": "octo-org/octo-repo"},
2821 ],
2822 });
2823 let normalized = connector_normalized(
2824 "installation_repositories",
2825 "delivery-inst-repos",
2826 3001,
2827 Some("removed"),
2828 original.clone(),
2829 serde_json::json!({
2830 "installation": original["installation"].clone(),
2831 "account": {"login": "octo-org"},
2832 "installation_state": "revoked",
2833 "suspended": false,
2834 "revoked": true,
2835 "repository_selection": "selected",
2836 "repositories_added": [],
2837 "repositories_removed": original["repositories_removed"].clone(),
2838 }),
2839 );
2840 let provider = ProviderId::from("github");
2841 let payload = ProviderPayload::normalize(
2842 &provider,
2843 "installation_repositories",
2844 &github_headers("installation_repositories", "delivery-inst-repos"),
2845 normalized,
2846 )
2847 .expect("installation_repositories payload");
2848 let GitHubEventPayload::InstallationRepositories(repos) = unwrap_github(payload) else {
2849 panic!("expected InstallationRepositories variant");
2850 };
2851 assert_eq!(repos.common.event, "installation_repositories");
2852 assert_eq!(repos.common.action.as_deref(), Some("removed"));
2853 assert_eq!(repos.repository_selection.as_deref(), Some("selected"));
2854 assert!(repos.repositories_added.is_empty());
2855 assert_eq!(repos.repositories_removed.len(), 1);
2856 assert_eq!(
2857 repos.repositories_removed[0]
2858 .get("full_name")
2859 .and_then(|v| v.as_str()),
2860 Some("octo-org/octo-repo"),
2861 );
2862 assert_eq!(repos.installation_state.as_deref(), Some("revoked"));
2863 assert_eq!(repos.revoked, Some(true));
2864 }
2865
2866 #[test]
2867 fn github_legacy_direct_webhook_still_normalizes() {
2868 let provider = ProviderId::from("github");
2872 let payload = ProviderPayload::normalize(
2873 &provider,
2874 "issues",
2875 &github_headers("issues", "delivery-legacy"),
2876 serde_json::json!({
2877 "action": "opened",
2878 "installation": {"id": 99},
2879 "issue": {"number": 7},
2880 }),
2881 )
2882 .expect("legacy issues payload");
2883 let GitHubEventPayload::Issues(issues) = unwrap_github(payload) else {
2884 panic!("expected Issues variant");
2885 };
2886 assert_eq!(issues.common.installation_id, Some(99));
2887 assert_eq!(
2888 issues.common.delivery_id.as_deref(),
2889 Some("delivery-legacy")
2890 );
2891 assert!(issues.common.topic.is_none());
2892 assert!(issues.common.repo.is_none());
2893 assert_eq!(issues.issue.get("number").and_then(|v| v.as_i64()), Some(7));
2894 }
2895
2896 #[test]
2897 fn github_new_event_variants_round_trip_through_serde() {
2898 let provider = ProviderId::from("github");
2903 let cases: &[(&str, serde_json::Value, &str)] = &[
2904 (
2905 "check_suite",
2906 serde_json::json!({
2907 "event": "check_suite",
2908 "check_suite": {"id": 1},
2909 "check_suite_id": 1,
2910 "raw": {"check_suite": {"id": 1}},
2911 }),
2912 "CheckSuite",
2913 ),
2914 (
2915 "status",
2916 serde_json::json!({
2917 "event": "status",
2918 "commit_status": {"id": 9},
2919 "status_id": 9,
2920 "state": "success",
2921 "raw": {"id": 9, "state": "success"},
2922 }),
2923 "Status",
2924 ),
2925 (
2926 "merge_group",
2927 serde_json::json!({
2928 "event": "merge_group",
2929 "merge_group": {"id": 1},
2930 "merge_group_id": 1,
2931 "raw": {"merge_group": {"id": 1}},
2932 }),
2933 "MergeGroup",
2934 ),
2935 (
2936 "installation",
2937 serde_json::json!({
2938 "event": "installation",
2939 "installation": {"id": 1},
2940 "installation_state": "active",
2941 "suspended": false,
2942 "raw": {"installation": {"id": 1}},
2943 }),
2944 "Installation",
2945 ),
2946 (
2947 "installation_repositories",
2948 serde_json::json!({
2949 "event": "installation_repositories",
2950 "installation": {"id": 1},
2951 "repository_selection": "selected",
2952 "repositories_added": [],
2953 "repositories_removed": [{"id": 7}],
2954 "raw": {"installation": {"id": 1}},
2955 }),
2956 "InstallationRepositories",
2957 ),
2958 ];
2959 for (kind, raw, want_variant) in cases {
2960 let payload = ProviderPayload::normalize(
2961 &provider,
2962 kind,
2963 &github_headers(kind, "delivery"),
2964 raw.clone(),
2965 )
2966 .unwrap_or_else(|_| panic!("normalize {kind}"));
2967 let serialized = serde_json::to_value(&payload).expect("serialize");
2968 let deserialized: ProviderPayload =
2969 serde_json::from_value(serialized.clone()).expect("deserialize");
2970 let actual_variant = match unwrap_github(deserialized) {
2971 GitHubEventPayload::Issues(_) => "Issues",
2972 GitHubEventPayload::PullRequest(_) => "PullRequest",
2973 GitHubEventPayload::IssueComment(_) => "IssueComment",
2974 GitHubEventPayload::PullRequestReview(_) => "PullRequestReview",
2975 GitHubEventPayload::Push(_) => "Push",
2976 GitHubEventPayload::WorkflowRun(_) => "WorkflowRun",
2977 GitHubEventPayload::DeploymentStatus(_) => "DeploymentStatus",
2978 GitHubEventPayload::CheckRun(_) => "CheckRun",
2979 GitHubEventPayload::CheckSuite(_) => "CheckSuite",
2980 GitHubEventPayload::Status(_) => "Status",
2981 GitHubEventPayload::MergeGroup(_) => "MergeGroup",
2982 GitHubEventPayload::Installation(_) => "Installation",
2983 GitHubEventPayload::InstallationRepositories(_) => "InstallationRepositories",
2984 GitHubEventPayload::Other(_) => "Other",
2985 };
2986 assert_eq!(
2987 actual_variant, *want_variant,
2988 "{kind} round-tripped as {actual_variant}, expected {want_variant}; serialized form: {serialized}"
2989 );
2990 }
2991 }
2992
2993 #[test]
2994 fn provider_normalizes_stream_payloads() {
2995 let payload = ProviderPayload::normalize(
2996 &ProviderId::from("kafka"),
2997 "quote.tick",
2998 &BTreeMap::from([("x-source".to_string(), "feed".to_string())]),
2999 serde_json::json!({
3000 "topic": "quotes",
3001 "partition": 7,
3002 "offset": "42",
3003 "key": "AAPL",
3004 "timestamp": "2026-04-21T12:00:00Z"
3005 }),
3006 )
3007 .expect("stream payload");
3008 let ProviderPayload::Known(KnownProviderPayload::Kafka(payload)) = payload else {
3009 panic!("expected kafka stream payload")
3010 };
3011 assert_eq!(payload.event, "quote.tick");
3012 assert_eq!(payload.stream.as_deref(), Some("quotes"));
3013 assert_eq!(payload.partition.as_deref(), Some("7"));
3014 assert_eq!(payload.offset.as_deref(), Some("42"));
3015 assert_eq!(payload.key.as_deref(), Some("AAPL"));
3016 assert_eq!(payload.timestamp.as_deref(), Some("2026-04-21T12:00:00Z"));
3017 }
3018}