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
9use crate::triggers::test_util::clock;
10
11const REDACTED_HEADER_VALUE: &str = "[redacted]";
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
905#[derive(Clone, Debug, PartialEq, Eq)]
906pub struct HeaderRedactionPolicy {
907 safe_exact_names: BTreeSet<String>,
908}
909
910impl HeaderRedactionPolicy {
911 pub fn with_safe_header(mut self, name: impl Into<String>) -> Self {
912 self.safe_exact_names
913 .insert(name.into().to_ascii_lowercase());
914 self
915 }
916
917 fn should_keep(&self, name: &str) -> bool {
918 let lower = name.to_ascii_lowercase();
919 if self.safe_exact_names.contains(lower.as_str()) {
920 return true;
921 }
922 matches!(
923 lower.as_str(),
924 "user-agent"
925 | "request-id"
926 | "x-request-id"
927 | "x-correlation-id"
928 | "content-type"
929 | "content-length"
930 | "x-github-event"
931 | "x-github-delivery"
932 | "x-github-hook-id"
933 | "x-hub-signature-256"
934 | "x-slack-request-timestamp"
935 | "x-slack-signature"
936 | "x-linear-signature"
937 | "x-notion-signature"
938 | "x-a2a-signature"
939 | "x-a2a-delivery"
940 ) || lower.ends_with("-event")
941 || lower.ends_with("-delivery")
942 || lower.contains("timestamp")
943 || lower.contains("request-id")
944 }
945
946 fn should_redact(&self, name: &str) -> bool {
947 let lower = name.to_ascii_lowercase();
948 if self.should_keep(lower.as_str()) {
949 return false;
950 }
951 lower.contains("authorization")
952 || lower.contains("cookie")
953 || lower.contains("secret")
954 || lower.contains("token")
955 || lower.contains("key")
956 }
957}
958
959impl Default for HeaderRedactionPolicy {
960 fn default() -> Self {
961 Self {
962 safe_exact_names: BTreeSet::from([
963 "content-length".to_string(),
964 "content-type".to_string(),
965 "request-id".to_string(),
966 "user-agent".to_string(),
967 "x-a2a-delivery".to_string(),
968 "x-a2a-signature".to_string(),
969 "x-correlation-id".to_string(),
970 "x-github-delivery".to_string(),
971 "x-github-event".to_string(),
972 "x-github-hook-id".to_string(),
973 "x-hub-signature-256".to_string(),
974 "x-linear-signature".to_string(),
975 "x-notion-signature".to_string(),
976 "x-request-id".to_string(),
977 "x-slack-request-timestamp".to_string(),
978 "x-slack-signature".to_string(),
979 ]),
980 }
981 }
982}
983
984pub fn redact_headers(
985 headers: &BTreeMap<String, String>,
986 policy: &HeaderRedactionPolicy,
987) -> BTreeMap<String, String> {
988 headers
989 .iter()
990 .map(|(name, value)| {
991 if policy.should_redact(name) {
992 (name.clone(), REDACTED_HEADER_VALUE.to_string())
993 } else {
994 (name.clone(), value.clone())
995 }
996 })
997 .collect()
998}
999
1000#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1001pub struct ProviderSecretRequirement {
1002 pub name: String,
1003 pub required: bool,
1004 pub namespace: String,
1005}
1006
1007#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1008pub struct ProviderOutboundMethod {
1009 pub name: String,
1010}
1011
1012#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
1013#[serde(tag = "kind", rename_all = "snake_case")]
1014pub enum SignatureVerificationMetadata {
1015 #[default]
1016 None,
1017 Hmac {
1018 variant: String,
1019 raw_body: bool,
1020 signature_header: String,
1021 timestamp_header: Option<String>,
1022 id_header: Option<String>,
1023 default_tolerance_secs: Option<i64>,
1024 digest: String,
1025 encoding: String,
1026 },
1027}
1028
1029#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
1030#[serde(tag = "kind", rename_all = "snake_case")]
1031pub enum ProviderRuntimeMetadata {
1032 Builtin {
1033 connector: String,
1034 default_signature_variant: Option<String>,
1035 },
1036 #[default]
1037 Placeholder,
1038}
1039
1040#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
1041pub struct ProviderMetadata {
1042 pub provider: String,
1043 #[serde(default)]
1044 pub kinds: Vec<String>,
1045 pub schema_name: String,
1046 #[serde(default)]
1047 pub outbound_methods: Vec<ProviderOutboundMethod>,
1048 #[serde(default)]
1049 pub secret_requirements: Vec<ProviderSecretRequirement>,
1050 #[serde(default)]
1051 pub signature_verification: SignatureVerificationMetadata,
1052 #[serde(default)]
1053 pub runtime: ProviderRuntimeMetadata,
1054}
1055
1056impl ProviderMetadata {
1057 pub fn supports_kind(&self, kind: &str) -> bool {
1058 self.kinds.iter().any(|candidate| candidate == kind)
1059 }
1060
1061 pub fn required_secret_names(&self) -> impl Iterator<Item = &str> {
1062 self.secret_requirements
1063 .iter()
1064 .filter(|requirement| requirement.required)
1065 .map(|requirement| requirement.name.as_str())
1066 }
1067}
1068
1069pub trait ProviderSchema: Send + Sync {
1070 fn provider_id(&self) -> &'static str;
1071 fn harn_schema_name(&self) -> &'static str;
1072 fn metadata(&self) -> ProviderMetadata {
1073 ProviderMetadata {
1074 provider: self.provider_id().to_string(),
1075 schema_name: self.harn_schema_name().to_string(),
1076 ..ProviderMetadata::default()
1077 }
1078 }
1079 fn normalize(
1080 &self,
1081 kind: &str,
1082 headers: &BTreeMap<String, String>,
1083 raw: JsonValue,
1084 ) -> Result<ProviderPayload, ProviderCatalogError>;
1085}
1086
1087#[derive(Clone, Debug, PartialEq, Eq)]
1088pub enum ProviderCatalogError {
1089 DuplicateProvider(String),
1090 UnknownProvider(String),
1091}
1092
1093impl std::fmt::Display for ProviderCatalogError {
1094 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1095 match self {
1096 Self::DuplicateProvider(provider) => {
1097 write!(f, "provider `{provider}` is already registered")
1098 }
1099 Self::UnknownProvider(provider) => write!(f, "provider `{provider}` is not registered"),
1100 }
1101 }
1102}
1103
1104impl std::error::Error for ProviderCatalogError {}
1105
1106#[derive(Clone, Default)]
1107pub struct ProviderCatalog {
1108 providers: BTreeMap<String, Arc<dyn ProviderSchema>>,
1109}
1110
1111impl ProviderCatalog {
1112 pub fn with_defaults() -> Self {
1113 let mut catalog = Self::default();
1114 for schema in default_provider_schemas() {
1115 catalog
1116 .register(schema)
1117 .expect("default providers must register cleanly");
1118 }
1119 catalog
1120 }
1121
1122 pub fn register(
1123 &mut self,
1124 schema: Arc<dyn ProviderSchema>,
1125 ) -> Result<(), ProviderCatalogError> {
1126 let provider = schema.provider_id().to_string();
1127 if self.providers.contains_key(provider.as_str()) {
1128 return Err(ProviderCatalogError::DuplicateProvider(provider));
1129 }
1130 self.providers.insert(provider, schema);
1131 Ok(())
1132 }
1133
1134 pub fn normalize(
1135 &self,
1136 provider: &ProviderId,
1137 kind: &str,
1138 headers: &BTreeMap<String, String>,
1139 raw: JsonValue,
1140 ) -> Result<ProviderPayload, ProviderCatalogError> {
1141 let schema = self
1142 .providers
1143 .get(provider.as_str())
1144 .ok_or_else(|| ProviderCatalogError::UnknownProvider(provider.0.clone()))?;
1145 schema.normalize(kind, headers, raw)
1146 }
1147
1148 pub fn schema_names(&self) -> BTreeMap<String, String> {
1149 self.providers
1150 .iter()
1151 .map(|(provider, schema)| (provider.clone(), schema.harn_schema_name().to_string()))
1152 .collect()
1153 }
1154
1155 pub fn entries(&self) -> Vec<ProviderMetadata> {
1156 self.providers
1157 .values()
1158 .map(|schema| schema.metadata())
1159 .collect()
1160 }
1161
1162 pub fn metadata_for(&self, provider: &str) -> Option<ProviderMetadata> {
1163 self.providers.get(provider).map(|schema| schema.metadata())
1164 }
1165}
1166
1167pub fn register_provider_schema(
1168 schema: Arc<dyn ProviderSchema>,
1169) -> Result<(), ProviderCatalogError> {
1170 provider_catalog()
1171 .write()
1172 .expect("provider catalog poisoned")
1173 .register(schema)
1174}
1175
1176pub fn reset_provider_catalog() {
1177 *provider_catalog()
1178 .write()
1179 .expect("provider catalog poisoned") = ProviderCatalog::with_defaults();
1180}
1181
1182pub fn reset_provider_catalog_with(
1183 schemas: Vec<Arc<dyn ProviderSchema>>,
1184) -> Result<(), ProviderCatalogError> {
1185 let mut catalog = ProviderCatalog::with_defaults();
1186 let builtin_providers: BTreeSet<String> = catalog.schema_names().into_keys().collect();
1187 for schema in schemas {
1188 if builtin_providers.contains(schema.provider_id()) {
1189 continue;
1190 }
1191 catalog.register(schema)?;
1192 }
1193 *provider_catalog()
1194 .write()
1195 .expect("provider catalog poisoned") = catalog;
1196 Ok(())
1197}
1198
1199pub fn registered_provider_schema_names() -> BTreeMap<String, String> {
1200 provider_catalog()
1201 .read()
1202 .expect("provider catalog poisoned")
1203 .schema_names()
1204}
1205
1206pub fn registered_provider_metadata() -> Vec<ProviderMetadata> {
1207 provider_catalog()
1208 .read()
1209 .expect("provider catalog poisoned")
1210 .entries()
1211}
1212
1213pub fn provider_metadata(provider: &str) -> Option<ProviderMetadata> {
1214 provider_catalog()
1215 .read()
1216 .expect("provider catalog poisoned")
1217 .metadata_for(provider)
1218}
1219
1220fn provider_catalog() -> &'static RwLock<ProviderCatalog> {
1221 static PROVIDER_CATALOG: OnceLock<RwLock<ProviderCatalog>> = OnceLock::new();
1222 PROVIDER_CATALOG.get_or_init(|| RwLock::new(ProviderCatalog::with_defaults()))
1223}
1224
1225struct BuiltinProviderSchema {
1226 provider_id: &'static str,
1227 harn_schema_name: &'static str,
1228 metadata: ProviderMetadata,
1229 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
1230}
1231
1232impl ProviderSchema for BuiltinProviderSchema {
1233 fn provider_id(&self) -> &'static str {
1234 self.provider_id
1235 }
1236
1237 fn harn_schema_name(&self) -> &'static str {
1238 self.harn_schema_name
1239 }
1240
1241 fn metadata(&self) -> ProviderMetadata {
1242 self.metadata.clone()
1243 }
1244
1245 fn normalize(
1246 &self,
1247 kind: &str,
1248 headers: &BTreeMap<String, String>,
1249 raw: JsonValue,
1250 ) -> Result<ProviderPayload, ProviderCatalogError> {
1251 Ok((self.normalize)(kind, headers, raw))
1252 }
1253}
1254
1255fn provider_metadata_entry(
1256 provider: &str,
1257 kinds: &[&str],
1258 schema_name: &str,
1259 outbound_methods: &[&str],
1260 signature_verification: SignatureVerificationMetadata,
1261 secret_requirements: Vec<ProviderSecretRequirement>,
1262 runtime: ProviderRuntimeMetadata,
1263) -> ProviderMetadata {
1264 ProviderMetadata {
1265 provider: provider.to_string(),
1266 kinds: kinds.iter().map(|kind| kind.to_string()).collect(),
1267 schema_name: schema_name.to_string(),
1268 outbound_methods: outbound_methods
1269 .iter()
1270 .map(|name| ProviderOutboundMethod {
1271 name: (*name).to_string(),
1272 })
1273 .collect(),
1274 secret_requirements,
1275 signature_verification,
1276 runtime,
1277 }
1278}
1279
1280fn hmac_signature_metadata(
1281 variant: &str,
1282 signature_header: &str,
1283 timestamp_header: Option<&str>,
1284 id_header: Option<&str>,
1285 default_tolerance_secs: Option<i64>,
1286 encoding: &str,
1287) -> SignatureVerificationMetadata {
1288 SignatureVerificationMetadata::Hmac {
1289 variant: variant.to_string(),
1290 raw_body: true,
1291 signature_header: signature_header.to_string(),
1292 timestamp_header: timestamp_header.map(ToString::to_string),
1293 id_header: id_header.map(ToString::to_string),
1294 default_tolerance_secs,
1295 digest: "sha256".to_string(),
1296 encoding: encoding.to_string(),
1297 }
1298}
1299
1300fn required_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
1301 ProviderSecretRequirement {
1302 name: name.to_string(),
1303 required: true,
1304 namespace: namespace.to_string(),
1305 }
1306}
1307
1308fn outbound_method(name: &str) -> ProviderOutboundMethod {
1309 ProviderOutboundMethod {
1310 name: name.to_string(),
1311 }
1312}
1313
1314fn optional_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
1315 ProviderSecretRequirement {
1316 name: name.to_string(),
1317 required: false,
1318 namespace: namespace.to_string(),
1319 }
1320}
1321
1322fn default_provider_schemas() -> Vec<Arc<dyn ProviderSchema>> {
1323 vec![
1324 Arc::new(BuiltinProviderSchema {
1325 provider_id: "github",
1326 harn_schema_name: "GitHubEventPayload",
1327 metadata: provider_metadata_entry(
1328 "github",
1329 &["webhook"],
1330 "GitHubEventPayload",
1331 &[],
1332 hmac_signature_metadata(
1333 "github",
1334 "X-Hub-Signature-256",
1335 None,
1336 Some("X-GitHub-Delivery"),
1337 None,
1338 "hex",
1339 ),
1340 vec![required_secret("signing_secret", "github")],
1341 ProviderRuntimeMetadata::Placeholder,
1342 ),
1343 normalize: github_payload,
1344 }),
1345 Arc::new(BuiltinProviderSchema {
1346 provider_id: "slack",
1347 harn_schema_name: "SlackEventPayload",
1348 metadata: provider_metadata_entry(
1349 "slack",
1350 &["webhook"],
1351 "SlackEventPayload",
1352 &[
1353 "post_message",
1354 "update_message",
1355 "add_reaction",
1356 "open_view",
1357 "user_info",
1358 "api_call",
1359 "upload_file",
1360 ],
1361 hmac_signature_metadata(
1362 "slack",
1363 "X-Slack-Signature",
1364 Some("X-Slack-Request-Timestamp"),
1365 None,
1366 Some(300),
1367 "hex",
1368 ),
1369 vec![required_secret("signing_secret", "slack")],
1370 ProviderRuntimeMetadata::Placeholder,
1371 ),
1372 normalize: slack_payload,
1373 }),
1374 Arc::new(BuiltinProviderSchema {
1375 provider_id: "linear",
1376 harn_schema_name: "LinearEventPayload",
1377 metadata: {
1378 let mut metadata = provider_metadata_entry(
1379 "linear",
1380 &["webhook"],
1381 "LinearEventPayload",
1382 &[],
1383 hmac_signature_metadata(
1384 "linear",
1385 "Linear-Signature",
1386 None,
1387 Some("Linear-Delivery"),
1388 Some(75),
1389 "hex",
1390 ),
1391 vec![
1392 required_secret("signing_secret", "linear"),
1393 optional_secret("access_token", "linear"),
1394 ],
1395 ProviderRuntimeMetadata::Placeholder,
1396 );
1397 metadata.outbound_methods = vec![
1398 ProviderOutboundMethod {
1399 name: "list_issues".to_string(),
1400 },
1401 ProviderOutboundMethod {
1402 name: "update_issue".to_string(),
1403 },
1404 ProviderOutboundMethod {
1405 name: "create_comment".to_string(),
1406 },
1407 ProviderOutboundMethod {
1408 name: "search".to_string(),
1409 },
1410 ProviderOutboundMethod {
1411 name: "graphql".to_string(),
1412 },
1413 ];
1414 metadata
1415 },
1416 normalize: linear_payload,
1417 }),
1418 Arc::new(BuiltinProviderSchema {
1419 provider_id: "notion",
1420 harn_schema_name: "NotionEventPayload",
1421 metadata: {
1422 let mut metadata = provider_metadata_entry(
1423 "notion",
1424 &["webhook", "poll"],
1425 "NotionEventPayload",
1426 &[],
1427 hmac_signature_metadata(
1428 "notion",
1429 "X-Notion-Signature",
1430 None,
1431 None,
1432 None,
1433 "hex",
1434 ),
1435 vec![required_secret("verification_token", "notion")],
1436 ProviderRuntimeMetadata::Placeholder,
1437 );
1438 metadata.outbound_methods = vec![
1439 outbound_method("get_page"),
1440 outbound_method("update_page"),
1441 outbound_method("append_blocks"),
1442 outbound_method("query_database"),
1443 outbound_method("search"),
1444 outbound_method("create_comment"),
1445 outbound_method("api_call"),
1446 ];
1447 metadata
1448 },
1449 normalize: notion_payload,
1450 }),
1451 Arc::new(BuiltinProviderSchema {
1452 provider_id: "cron",
1453 harn_schema_name: "CronEventPayload",
1454 metadata: provider_metadata_entry(
1455 "cron",
1456 &["cron"],
1457 "CronEventPayload",
1458 &[],
1459 SignatureVerificationMetadata::None,
1460 Vec::new(),
1461 ProviderRuntimeMetadata::Builtin {
1462 connector: "cron".to_string(),
1463 default_signature_variant: None,
1464 },
1465 ),
1466 normalize: cron_payload,
1467 }),
1468 Arc::new(BuiltinProviderSchema {
1469 provider_id: "webhook",
1470 harn_schema_name: "GenericWebhookPayload",
1471 metadata: provider_metadata_entry(
1472 "webhook",
1473 &["webhook"],
1474 "GenericWebhookPayload",
1475 &[],
1476 hmac_signature_metadata(
1477 "standard",
1478 "webhook-signature",
1479 Some("webhook-timestamp"),
1480 Some("webhook-id"),
1481 Some(300),
1482 "base64",
1483 ),
1484 vec![required_secret("signing_secret", "webhook")],
1485 ProviderRuntimeMetadata::Builtin {
1486 connector: "webhook".to_string(),
1487 default_signature_variant: Some("standard".to_string()),
1488 },
1489 ),
1490 normalize: webhook_payload,
1491 }),
1492 Arc::new(BuiltinProviderSchema {
1493 provider_id: "a2a-push",
1494 harn_schema_name: "A2aPushPayload",
1495 metadata: provider_metadata_entry(
1496 "a2a-push",
1497 &["a2a-push"],
1498 "A2aPushPayload",
1499 &[],
1500 SignatureVerificationMetadata::None,
1501 Vec::new(),
1502 ProviderRuntimeMetadata::Builtin {
1503 connector: "a2a-push".to_string(),
1504 default_signature_variant: None,
1505 },
1506 ),
1507 normalize: a2a_push_payload,
1508 }),
1509 Arc::new(stream_provider_schema("kafka", kafka_payload)),
1510 Arc::new(stream_provider_schema("nats", nats_payload)),
1511 Arc::new(stream_provider_schema("pulsar", pulsar_payload)),
1512 Arc::new(stream_provider_schema("postgres-cdc", postgres_cdc_payload)),
1513 Arc::new(stream_provider_schema("email", email_payload)),
1514 Arc::new(stream_provider_schema("websocket", websocket_payload)),
1515 ]
1516}
1517
1518fn stream_provider_schema(
1519 provider_id: &'static str,
1520 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
1521) -> BuiltinProviderSchema {
1522 BuiltinProviderSchema {
1523 provider_id,
1524 harn_schema_name: "StreamEventPayload",
1525 metadata: provider_metadata_entry(
1526 provider_id,
1527 &["stream"],
1528 "StreamEventPayload",
1529 &[],
1530 SignatureVerificationMetadata::None,
1531 Vec::new(),
1532 ProviderRuntimeMetadata::Builtin {
1533 connector: "stream".to_string(),
1534 default_signature_variant: None,
1535 },
1536 ),
1537 normalize,
1538 }
1539}
1540
1541fn github_payload(
1542 kind: &str,
1543 headers: &BTreeMap<String, String>,
1544 raw: JsonValue,
1545) -> ProviderPayload {
1546 let original_raw = raw
1551 .get("raw")
1552 .filter(|value| value.is_object())
1553 .cloned()
1554 .unwrap_or_else(|| raw.clone());
1555 let common = GitHubEventCommon {
1556 event: kind.to_string(),
1557 action: raw
1558 .get("action")
1559 .and_then(JsonValue::as_str)
1560 .map(ToString::to_string),
1561 delivery_id: raw
1562 .get("delivery_id")
1563 .and_then(JsonValue::as_str)
1564 .map(ToString::to_string)
1565 .or_else(|| headers.get("X-GitHub-Delivery").cloned()),
1566 installation_id: raw
1567 .get("installation_id")
1568 .and_then(JsonValue::as_i64)
1569 .or_else(|| {
1570 raw.get("installation")
1571 .and_then(|value| value.get("id"))
1572 .and_then(JsonValue::as_i64)
1573 }),
1574 topic: raw
1575 .get("topic")
1576 .and_then(JsonValue::as_str)
1577 .map(ToString::to_string),
1578 repository: raw.get("repository").cloned(),
1579 repo: raw.get("repo").cloned(),
1580 raw: original_raw,
1581 };
1582 let payload = match kind {
1583 "issues" => GitHubEventPayload::Issues(GitHubIssuesEventPayload {
1584 common,
1585 issue: raw.get("issue").cloned().unwrap_or(JsonValue::Null),
1586 }),
1587 "pull_request" => GitHubEventPayload::PullRequest(GitHubPullRequestEventPayload {
1588 common,
1589 pull_request: raw.get("pull_request").cloned().unwrap_or(JsonValue::Null),
1590 }),
1591 "issue_comment" => GitHubEventPayload::IssueComment(GitHubIssueCommentEventPayload {
1592 common,
1593 issue: raw.get("issue").cloned().unwrap_or(JsonValue::Null),
1594 comment: raw.get("comment").cloned().unwrap_or(JsonValue::Null),
1595 }),
1596 "pull_request_review" => {
1597 GitHubEventPayload::PullRequestReview(GitHubPullRequestReviewEventPayload {
1598 common,
1599 pull_request: raw.get("pull_request").cloned().unwrap_or(JsonValue::Null),
1600 review: raw.get("review").cloned().unwrap_or(JsonValue::Null),
1601 })
1602 }
1603 "push" => GitHubEventPayload::Push(GitHubPushEventPayload {
1604 common,
1605 commits: raw
1606 .get("commits")
1607 .and_then(JsonValue::as_array)
1608 .cloned()
1609 .unwrap_or_default(),
1610 distinct_size: raw.get("distinct_size").and_then(JsonValue::as_i64),
1611 }),
1612 "workflow_run" => GitHubEventPayload::WorkflowRun(GitHubWorkflowRunEventPayload {
1613 common,
1614 workflow_run: raw.get("workflow_run").cloned().unwrap_or(JsonValue::Null),
1615 }),
1616 "deployment_status" => {
1617 GitHubEventPayload::DeploymentStatus(GitHubDeploymentStatusEventPayload {
1618 common,
1619 deployment_status: raw
1620 .get("deployment_status")
1621 .cloned()
1622 .unwrap_or(JsonValue::Null),
1623 deployment: raw.get("deployment").cloned().unwrap_or(JsonValue::Null),
1624 })
1625 }
1626 "check_run" => GitHubEventPayload::CheckRun(GitHubCheckRunEventPayload {
1627 common,
1628 check_run: raw.get("check_run").cloned().unwrap_or(JsonValue::Null),
1629 }),
1630 "check_suite" => {
1631 let check_suite = raw.get("check_suite").cloned().unwrap_or(JsonValue::Null);
1632 GitHubEventPayload::CheckSuite(GitHubCheckSuiteEventPayload {
1633 check_suite_id: github_promoted_i64(&raw, "check_suite_id")
1634 .or_else(|| check_suite.get("id").and_then(JsonValue::as_i64)),
1635 pull_request_number: github_promoted_i64(&raw, "pull_request_number"),
1636 head_sha: github_promoted_string(&raw, "head_sha"),
1637 head_ref: github_promoted_string(&raw, "head_ref"),
1638 base_ref: github_promoted_string(&raw, "base_ref"),
1639 status: github_promoted_string(&raw, "status"),
1640 conclusion: github_promoted_string(&raw, "conclusion"),
1641 common,
1642 check_suite,
1643 })
1644 }
1645 "status" => GitHubEventPayload::Status(GitHubStatusEventPayload {
1646 commit_status: raw
1647 .get("commit_status")
1648 .cloned()
1649 .or_else(|| Some(common.raw.clone())),
1650 status_id: github_promoted_i64(&raw, "status_id")
1651 .or_else(|| common.raw.get("id").and_then(JsonValue::as_i64)),
1652 head_sha: github_promoted_string(&raw, "head_sha").or_else(|| {
1653 common
1654 .raw
1655 .get("sha")
1656 .and_then(JsonValue::as_str)
1657 .map(ToString::to_string)
1658 }),
1659 head_ref: github_promoted_string(&raw, "head_ref"),
1660 base_ref: github_promoted_string(&raw, "base_ref"),
1661 state: github_promoted_string(&raw, "state"),
1662 context: github_promoted_string(&raw, "context"),
1663 target_url: github_promoted_string(&raw, "target_url"),
1664 common,
1665 }),
1666 "merge_group" => {
1667 let merge_group = raw.get("merge_group").cloned().unwrap_or(JsonValue::Null);
1668 GitHubEventPayload::MergeGroup(GitHubMergeGroupEventPayload {
1669 merge_group_id: raw
1670 .get("merge_group_id")
1671 .cloned()
1672 .or_else(|| merge_group.get("id").cloned()),
1673 head_sha: github_promoted_string(&raw, "head_sha").or_else(|| {
1674 merge_group
1675 .get("head_sha")
1676 .and_then(JsonValue::as_str)
1677 .map(ToString::to_string)
1678 }),
1679 head_ref: github_promoted_string(&raw, "head_ref").or_else(|| {
1680 merge_group
1681 .get("head_ref")
1682 .and_then(JsonValue::as_str)
1683 .map(ToString::to_string)
1684 }),
1685 base_sha: github_promoted_string(&raw, "base_sha").or_else(|| {
1686 merge_group
1687 .get("base_sha")
1688 .and_then(JsonValue::as_str)
1689 .map(ToString::to_string)
1690 }),
1691 base_ref: github_promoted_string(&raw, "base_ref").or_else(|| {
1692 merge_group
1693 .get("base_ref")
1694 .and_then(JsonValue::as_str)
1695 .map(ToString::to_string)
1696 }),
1697 pull_requests: raw
1698 .get("pull_requests")
1699 .and_then(JsonValue::as_array)
1700 .cloned()
1701 .unwrap_or_default(),
1702 pull_request_numbers: raw
1703 .get("pull_request_numbers")
1704 .and_then(JsonValue::as_array)
1705 .map(|values| {
1706 values
1707 .iter()
1708 .filter_map(JsonValue::as_i64)
1709 .collect::<Vec<_>>()
1710 })
1711 .unwrap_or_default(),
1712 common,
1713 merge_group,
1714 })
1715 }
1716 "installation" => GitHubEventPayload::Installation(GitHubInstallationEventPayload {
1717 installation: raw.get("installation").cloned(),
1718 account: raw.get("account").cloned(),
1719 installation_state: github_promoted_string(&raw, "installation_state"),
1720 suspended: raw.get("suspended").and_then(JsonValue::as_bool),
1721 revoked: raw.get("revoked").and_then(JsonValue::as_bool),
1722 repositories: raw
1723 .get("repositories")
1724 .and_then(JsonValue::as_array)
1725 .cloned()
1726 .unwrap_or_default(),
1727 common,
1728 }),
1729 "installation_repositories" => GitHubEventPayload::InstallationRepositories(
1730 GitHubInstallationRepositoriesEventPayload {
1731 installation: raw.get("installation").cloned(),
1732 account: raw.get("account").cloned(),
1733 installation_state: github_promoted_string(&raw, "installation_state"),
1734 suspended: raw.get("suspended").and_then(JsonValue::as_bool),
1735 revoked: raw.get("revoked").and_then(JsonValue::as_bool),
1736 repository_selection: github_promoted_string(&raw, "repository_selection"),
1737 repositories_added: raw
1738 .get("repositories_added")
1739 .and_then(JsonValue::as_array)
1740 .cloned()
1741 .unwrap_or_default(),
1742 repositories_removed: raw
1743 .get("repositories_removed")
1744 .and_then(JsonValue::as_array)
1745 .cloned()
1746 .unwrap_or_default(),
1747 common,
1748 },
1749 ),
1750 _ => GitHubEventPayload::Other(common),
1751 };
1752 ProviderPayload::Known(KnownProviderPayload::GitHub(payload))
1753}
1754
1755fn github_promoted_string(raw: &JsonValue, field: &str) -> Option<String> {
1756 raw.get(field)
1757 .and_then(JsonValue::as_str)
1758 .map(ToString::to_string)
1759}
1760
1761fn github_promoted_i64(raw: &JsonValue, field: &str) -> Option<i64> {
1762 raw.get(field).and_then(JsonValue::as_i64)
1763}
1764
1765fn slack_payload(
1766 kind: &str,
1767 _headers: &BTreeMap<String, String>,
1768 raw: JsonValue,
1769) -> ProviderPayload {
1770 let event = raw.get("event");
1771 let common = SlackEventCommon {
1772 event: kind.to_string(),
1773 event_id: raw
1774 .get("event_id")
1775 .and_then(JsonValue::as_str)
1776 .map(ToString::to_string),
1777 api_app_id: raw
1778 .get("api_app_id")
1779 .and_then(JsonValue::as_str)
1780 .map(ToString::to_string),
1781 team_id: raw
1782 .get("team_id")
1783 .and_then(JsonValue::as_str)
1784 .map(ToString::to_string),
1785 channel_id: slack_channel_id(event),
1786 user_id: slack_user_id(event),
1787 event_ts: event
1788 .and_then(|value| value.get("event_ts"))
1789 .and_then(JsonValue::as_str)
1790 .map(ToString::to_string),
1791 raw: raw.clone(),
1792 };
1793 let payload = match kind {
1794 kind if kind == "message" || kind.starts_with("message.") => {
1795 SlackEventPayload::Message(SlackMessageEventPayload {
1796 subtype: event
1797 .and_then(|value| value.get("subtype"))
1798 .and_then(JsonValue::as_str)
1799 .map(ToString::to_string),
1800 channel_type: event
1801 .and_then(|value| value.get("channel_type"))
1802 .and_then(JsonValue::as_str)
1803 .map(ToString::to_string),
1804 channel: event
1805 .and_then(|value| value.get("channel"))
1806 .and_then(JsonValue::as_str)
1807 .map(ToString::to_string),
1808 user: event
1809 .and_then(|value| value.get("user"))
1810 .and_then(JsonValue::as_str)
1811 .map(ToString::to_string),
1812 text: event
1813 .and_then(|value| value.get("text"))
1814 .and_then(JsonValue::as_str)
1815 .map(ToString::to_string),
1816 ts: event
1817 .and_then(|value| value.get("ts"))
1818 .and_then(JsonValue::as_str)
1819 .map(ToString::to_string),
1820 thread_ts: event
1821 .and_then(|value| value.get("thread_ts"))
1822 .and_then(JsonValue::as_str)
1823 .map(ToString::to_string),
1824 common,
1825 })
1826 }
1827 "app_mention" => SlackEventPayload::AppMention(SlackAppMentionEventPayload {
1828 channel: event
1829 .and_then(|value| value.get("channel"))
1830 .and_then(JsonValue::as_str)
1831 .map(ToString::to_string),
1832 user: event
1833 .and_then(|value| value.get("user"))
1834 .and_then(JsonValue::as_str)
1835 .map(ToString::to_string),
1836 text: event
1837 .and_then(|value| value.get("text"))
1838 .and_then(JsonValue::as_str)
1839 .map(ToString::to_string),
1840 ts: event
1841 .and_then(|value| value.get("ts"))
1842 .and_then(JsonValue::as_str)
1843 .map(ToString::to_string),
1844 thread_ts: event
1845 .and_then(|value| value.get("thread_ts"))
1846 .and_then(JsonValue::as_str)
1847 .map(ToString::to_string),
1848 common,
1849 }),
1850 "reaction_added" => SlackEventPayload::ReactionAdded(SlackReactionAddedEventPayload {
1851 reaction: event
1852 .and_then(|value| value.get("reaction"))
1853 .and_then(JsonValue::as_str)
1854 .map(ToString::to_string),
1855 item_user: event
1856 .and_then(|value| value.get("item_user"))
1857 .and_then(JsonValue::as_str)
1858 .map(ToString::to_string),
1859 item: event
1860 .and_then(|value| value.get("item"))
1861 .cloned()
1862 .unwrap_or(JsonValue::Null),
1863 common,
1864 }),
1865 "app_home_opened" => SlackEventPayload::AppHomeOpened(SlackAppHomeOpenedEventPayload {
1866 user: event
1867 .and_then(|value| value.get("user"))
1868 .and_then(JsonValue::as_str)
1869 .map(ToString::to_string),
1870 channel: event
1871 .and_then(|value| value.get("channel"))
1872 .and_then(JsonValue::as_str)
1873 .map(ToString::to_string),
1874 tab: event
1875 .and_then(|value| value.get("tab"))
1876 .and_then(JsonValue::as_str)
1877 .map(ToString::to_string),
1878 view: event
1879 .and_then(|value| value.get("view"))
1880 .cloned()
1881 .unwrap_or(JsonValue::Null),
1882 common,
1883 }),
1884 "assistant_thread_started" => {
1885 let assistant_thread = event
1886 .and_then(|value| value.get("assistant_thread"))
1887 .cloned()
1888 .unwrap_or(JsonValue::Null);
1889 SlackEventPayload::AssistantThreadStarted(SlackAssistantThreadStartedEventPayload {
1890 thread_ts: assistant_thread
1891 .get("thread_ts")
1892 .and_then(JsonValue::as_str)
1893 .map(ToString::to_string),
1894 context: assistant_thread
1895 .get("context")
1896 .cloned()
1897 .unwrap_or(JsonValue::Null),
1898 assistant_thread,
1899 common,
1900 })
1901 }
1902 _ => SlackEventPayload::Other(common),
1903 };
1904 ProviderPayload::Known(KnownProviderPayload::Slack(Box::new(payload)))
1905}
1906
1907fn slack_channel_id(event: Option<&JsonValue>) -> Option<String> {
1908 event
1909 .and_then(|value| value.get("channel"))
1910 .and_then(JsonValue::as_str)
1911 .map(ToString::to_string)
1912 .or_else(|| {
1913 event
1914 .and_then(|value| value.get("item"))
1915 .and_then(|value| value.get("channel"))
1916 .and_then(JsonValue::as_str)
1917 .map(ToString::to_string)
1918 })
1919 .or_else(|| {
1920 event
1921 .and_then(|value| value.get("channel"))
1922 .and_then(|value| value.get("id"))
1923 .and_then(JsonValue::as_str)
1924 .map(ToString::to_string)
1925 })
1926 .or_else(|| {
1927 event
1928 .and_then(|value| value.get("assistant_thread"))
1929 .and_then(|value| value.get("channel_id"))
1930 .and_then(JsonValue::as_str)
1931 .map(ToString::to_string)
1932 })
1933}
1934
1935fn slack_user_id(event: Option<&JsonValue>) -> Option<String> {
1936 event
1937 .and_then(|value| value.get("user"))
1938 .and_then(JsonValue::as_str)
1939 .map(ToString::to_string)
1940 .or_else(|| {
1941 event
1942 .and_then(|value| value.get("user"))
1943 .and_then(|value| value.get("id"))
1944 .and_then(JsonValue::as_str)
1945 .map(ToString::to_string)
1946 })
1947 .or_else(|| {
1948 event
1949 .and_then(|value| value.get("item_user"))
1950 .and_then(JsonValue::as_str)
1951 .map(ToString::to_string)
1952 })
1953 .or_else(|| {
1954 event
1955 .and_then(|value| value.get("assistant_thread"))
1956 .and_then(|value| value.get("user_id"))
1957 .and_then(JsonValue::as_str)
1958 .map(ToString::to_string)
1959 })
1960}
1961
1962fn linear_payload(
1963 _kind: &str,
1964 headers: &BTreeMap<String, String>,
1965 raw: JsonValue,
1966) -> ProviderPayload {
1967 let common = linear_event_common(headers, &raw);
1968 let event = common.event.clone();
1969 let payload = match event.as_str() {
1970 "issue" => LinearEventPayload::Issue(LinearIssueEventPayload {
1971 common,
1972 issue: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1973 changes: parse_linear_issue_changes(raw.get("updatedFrom")),
1974 }),
1975 "comment" => LinearEventPayload::IssueComment(LinearIssueCommentEventPayload {
1976 common,
1977 comment: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1978 }),
1979 "issue_label" => LinearEventPayload::IssueLabel(LinearIssueLabelEventPayload {
1980 common,
1981 label: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1982 }),
1983 "project" => LinearEventPayload::Project(LinearProjectEventPayload {
1984 common,
1985 project: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1986 }),
1987 "cycle" => LinearEventPayload::Cycle(LinearCycleEventPayload {
1988 common,
1989 cycle: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1990 }),
1991 "customer" => LinearEventPayload::Customer(LinearCustomerEventPayload {
1992 common,
1993 customer: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1994 }),
1995 "customer_request" => {
1996 LinearEventPayload::CustomerRequest(LinearCustomerRequestEventPayload {
1997 common,
1998 customer_request: raw.get("data").cloned().unwrap_or(JsonValue::Null),
1999 })
2000 }
2001 _ => LinearEventPayload::Other(common),
2002 };
2003 ProviderPayload::Known(KnownProviderPayload::Linear(payload))
2004}
2005
2006fn linear_event_common(headers: &BTreeMap<String, String>, raw: &JsonValue) -> LinearEventCommon {
2007 LinearEventCommon {
2008 event: linear_event_name(
2009 raw.get("type")
2010 .and_then(JsonValue::as_str)
2011 .or_else(|| headers.get("Linear-Event").map(String::as_str)),
2012 ),
2013 action: raw
2014 .get("action")
2015 .and_then(JsonValue::as_str)
2016 .map(ToString::to_string),
2017 delivery_id: header_value(headers, "Linear-Delivery").map(ToString::to_string),
2018 organization_id: raw
2019 .get("organizationId")
2020 .and_then(JsonValue::as_str)
2021 .map(ToString::to_string),
2022 webhook_timestamp: raw.get("webhookTimestamp").and_then(parse_json_i64ish),
2023 webhook_id: raw
2024 .get("webhookId")
2025 .and_then(JsonValue::as_str)
2026 .map(ToString::to_string),
2027 url: raw
2028 .get("url")
2029 .and_then(JsonValue::as_str)
2030 .map(ToString::to_string),
2031 created_at: raw
2032 .get("createdAt")
2033 .and_then(JsonValue::as_str)
2034 .map(ToString::to_string),
2035 actor: raw.get("actor").cloned().unwrap_or(JsonValue::Null),
2036 raw: raw.clone(),
2037 }
2038}
2039
2040fn linear_event_name(raw_type: Option<&str>) -> String {
2041 match raw_type.unwrap_or_default().to_ascii_lowercase().as_str() {
2042 "issue" => "issue".to_string(),
2043 "comment" | "issuecomment" | "issue_comment" => "comment".to_string(),
2044 "issuelabel" | "issue_label" => "issue_label".to_string(),
2045 "project" | "projectupdate" | "project_update" => "project".to_string(),
2046 "cycle" => "cycle".to_string(),
2047 "customer" => "customer".to_string(),
2048 "customerrequest" | "customer_request" => "customer_request".to_string(),
2049 other if !other.is_empty() => other.to_string(),
2050 _ => "other".to_string(),
2051 }
2052}
2053
2054fn parse_linear_issue_changes(updated_from: Option<&JsonValue>) -> Vec<LinearIssueChange> {
2055 let Some(JsonValue::Object(fields)) = updated_from else {
2056 return Vec::new();
2057 };
2058 let mut changes = Vec::new();
2059 for (field, previous) in fields {
2060 let change = match field.as_str() {
2061 "title" => LinearIssueChange::Title {
2062 previous: previous.as_str().map(ToString::to_string),
2063 },
2064 "description" => LinearIssueChange::Description {
2065 previous: previous.as_str().map(ToString::to_string),
2066 },
2067 "priority" => LinearIssueChange::Priority {
2068 previous: parse_json_i64ish(previous),
2069 },
2070 "estimate" => LinearIssueChange::Estimate {
2071 previous: parse_json_i64ish(previous),
2072 },
2073 "stateId" => LinearIssueChange::StateId {
2074 previous: previous.as_str().map(ToString::to_string),
2075 },
2076 "teamId" => LinearIssueChange::TeamId {
2077 previous: previous.as_str().map(ToString::to_string),
2078 },
2079 "assigneeId" => LinearIssueChange::AssigneeId {
2080 previous: previous.as_str().map(ToString::to_string),
2081 },
2082 "projectId" => LinearIssueChange::ProjectId {
2083 previous: previous.as_str().map(ToString::to_string),
2084 },
2085 "cycleId" => LinearIssueChange::CycleId {
2086 previous: previous.as_str().map(ToString::to_string),
2087 },
2088 "dueDate" => LinearIssueChange::DueDate {
2089 previous: previous.as_str().map(ToString::to_string),
2090 },
2091 "parentId" => LinearIssueChange::ParentId {
2092 previous: previous.as_str().map(ToString::to_string),
2093 },
2094 "sortOrder" => LinearIssueChange::SortOrder {
2095 previous: previous.as_f64(),
2096 },
2097 "labelIds" => LinearIssueChange::LabelIds {
2098 previous: parse_string_array(previous),
2099 },
2100 "completedAt" => LinearIssueChange::CompletedAt {
2101 previous: previous.as_str().map(ToString::to_string),
2102 },
2103 _ => LinearIssueChange::Other {
2104 field: field.clone(),
2105 previous: previous.clone(),
2106 },
2107 };
2108 changes.push(change);
2109 }
2110 changes
2111}
2112
2113fn parse_json_i64ish(value: &JsonValue) -> Option<i64> {
2114 value
2115 .as_i64()
2116 .or_else(|| value.as_u64().and_then(|raw| i64::try_from(raw).ok()))
2117 .or_else(|| value.as_str().and_then(|raw| raw.parse::<i64>().ok()))
2118}
2119
2120fn parse_string_array(value: &JsonValue) -> Vec<String> {
2121 let Some(array) = value.as_array() else {
2122 return Vec::new();
2123 };
2124 array
2125 .iter()
2126 .filter_map(|entry| {
2127 entry.as_str().map(ToString::to_string).or_else(|| {
2128 entry
2129 .get("id")
2130 .and_then(JsonValue::as_str)
2131 .map(ToString::to_string)
2132 })
2133 })
2134 .collect()
2135}
2136
2137fn header_value<'a>(headers: &'a BTreeMap<String, String>, name: &str) -> Option<&'a str> {
2138 headers
2139 .iter()
2140 .find(|(key, _)| key.eq_ignore_ascii_case(name))
2141 .map(|(_, value)| value.as_str())
2142}
2143
2144fn notion_payload(
2145 kind: &str,
2146 headers: &BTreeMap<String, String>,
2147 raw: JsonValue,
2148) -> ProviderPayload {
2149 let workspace_id = raw
2150 .get("workspace_id")
2151 .and_then(JsonValue::as_str)
2152 .map(ToString::to_string);
2153 ProviderPayload::Known(KnownProviderPayload::Notion(Box::new(NotionEventPayload {
2154 event: kind.to_string(),
2155 workspace_id,
2156 request_id: headers
2157 .get("request-id")
2158 .cloned()
2159 .or_else(|| headers.get("x-request-id").cloned()),
2160 subscription_id: raw
2161 .get("subscription_id")
2162 .and_then(JsonValue::as_str)
2163 .map(ToString::to_string),
2164 integration_id: raw
2165 .get("integration_id")
2166 .and_then(JsonValue::as_str)
2167 .map(ToString::to_string),
2168 attempt_number: raw
2169 .get("attempt_number")
2170 .and_then(JsonValue::as_u64)
2171 .and_then(|value| u32::try_from(value).ok()),
2172 entity_id: raw
2173 .get("entity")
2174 .and_then(|value| value.get("id"))
2175 .and_then(JsonValue::as_str)
2176 .map(ToString::to_string),
2177 entity_type: raw
2178 .get("entity")
2179 .and_then(|value| value.get("type"))
2180 .and_then(JsonValue::as_str)
2181 .map(ToString::to_string),
2182 api_version: raw
2183 .get("api_version")
2184 .and_then(JsonValue::as_str)
2185 .map(ToString::to_string),
2186 verification_token: raw
2187 .get("verification_token")
2188 .and_then(JsonValue::as_str)
2189 .map(ToString::to_string),
2190 polled: None,
2191 raw,
2192 })))
2193}
2194
2195fn cron_payload(
2196 _kind: &str,
2197 _headers: &BTreeMap<String, String>,
2198 raw: JsonValue,
2199) -> ProviderPayload {
2200 let cron_id = raw
2201 .get("cron_id")
2202 .and_then(JsonValue::as_str)
2203 .map(ToString::to_string);
2204 let schedule = raw
2205 .get("schedule")
2206 .and_then(JsonValue::as_str)
2207 .map(ToString::to_string);
2208 let tick_at = raw
2209 .get("tick_at")
2210 .and_then(JsonValue::as_str)
2211 .and_then(parse_rfc3339)
2212 .unwrap_or_else(OffsetDateTime::now_utc);
2213 ProviderPayload::Known(KnownProviderPayload::Cron(CronEventPayload {
2214 cron_id,
2215 schedule,
2216 tick_at,
2217 raw,
2218 }))
2219}
2220
2221fn webhook_payload(
2222 _kind: &str,
2223 headers: &BTreeMap<String, String>,
2224 raw: JsonValue,
2225) -> ProviderPayload {
2226 ProviderPayload::Known(KnownProviderPayload::Webhook(GenericWebhookPayload {
2227 source: headers.get("X-Webhook-Source").cloned(),
2228 content_type: headers.get("Content-Type").cloned(),
2229 raw,
2230 }))
2231}
2232
2233fn a2a_push_payload(
2234 _kind: &str,
2235 _headers: &BTreeMap<String, String>,
2236 raw: JsonValue,
2237) -> ProviderPayload {
2238 let task_id = raw
2239 .get("task_id")
2240 .and_then(JsonValue::as_str)
2241 .map(ToString::to_string);
2242 let sender = raw
2243 .get("sender")
2244 .and_then(JsonValue::as_str)
2245 .map(ToString::to_string);
2246 let task_state = raw
2247 .pointer("/status/state")
2248 .or_else(|| raw.pointer("/statusUpdate/status/state"))
2249 .and_then(JsonValue::as_str)
2250 .map(|state| match state {
2251 "cancelled" => "canceled".to_string(),
2252 other => other.to_string(),
2253 });
2254 let artifact = raw
2255 .pointer("/artifactUpdate/artifact")
2256 .or_else(|| raw.get("artifact"))
2257 .cloned();
2258 let kind = task_state
2259 .as_deref()
2260 .map(|state| format!("a2a.task.{state}"))
2261 .unwrap_or_else(|| "a2a.task.update".to_string());
2262 ProviderPayload::Known(KnownProviderPayload::A2aPush(A2aPushPayload {
2263 task_id,
2264 task_state,
2265 artifact,
2266 sender,
2267 raw,
2268 kind,
2269 }))
2270}
2271
2272fn kafka_payload(
2273 kind: &str,
2274 headers: &BTreeMap<String, String>,
2275 raw: JsonValue,
2276) -> ProviderPayload {
2277 ProviderPayload::Known(KnownProviderPayload::Kafka(stream_payload(
2278 kind, headers, raw,
2279 )))
2280}
2281
2282fn nats_payload(kind: &str, headers: &BTreeMap<String, String>, raw: JsonValue) -> ProviderPayload {
2283 ProviderPayload::Known(KnownProviderPayload::Nats(stream_payload(
2284 kind, headers, raw,
2285 )))
2286}
2287
2288fn pulsar_payload(
2289 kind: &str,
2290 headers: &BTreeMap<String, String>,
2291 raw: JsonValue,
2292) -> ProviderPayload {
2293 ProviderPayload::Known(KnownProviderPayload::Pulsar(stream_payload(
2294 kind, headers, raw,
2295 )))
2296}
2297
2298fn postgres_cdc_payload(
2299 kind: &str,
2300 headers: &BTreeMap<String, String>,
2301 raw: JsonValue,
2302) -> ProviderPayload {
2303 ProviderPayload::Known(KnownProviderPayload::PostgresCdc(stream_payload(
2304 kind, headers, raw,
2305 )))
2306}
2307
2308fn email_payload(
2309 kind: &str,
2310 headers: &BTreeMap<String, String>,
2311 raw: JsonValue,
2312) -> ProviderPayload {
2313 ProviderPayload::Known(KnownProviderPayload::Email(stream_payload(
2314 kind, headers, raw,
2315 )))
2316}
2317
2318fn websocket_payload(
2319 kind: &str,
2320 headers: &BTreeMap<String, String>,
2321 raw: JsonValue,
2322) -> ProviderPayload {
2323 ProviderPayload::Known(KnownProviderPayload::Websocket(stream_payload(
2324 kind, headers, raw,
2325 )))
2326}
2327
2328fn stream_payload(
2329 kind: &str,
2330 headers: &BTreeMap<String, String>,
2331 raw: JsonValue,
2332) -> StreamEventPayload {
2333 StreamEventPayload {
2334 event: kind.to_string(),
2335 source: json_stringish(&raw, &["source", "connector", "origin"]),
2336 stream: json_stringish(
2337 &raw,
2338 &["stream", "topic", "subject", "channel", "mailbox", "slot"],
2339 ),
2340 partition: json_stringish(&raw, &["partition", "shard", "consumer"]),
2341 offset: json_stringish(&raw, &["offset", "sequence", "lsn", "message_id"]),
2342 key: json_stringish(&raw, &["key", "message_key", "id", "event_id"]),
2343 timestamp: json_stringish(&raw, &["timestamp", "occurred_at", "received_at", "ts"]),
2344 headers: headers.clone(),
2345 raw,
2346 }
2347}
2348
2349fn json_stringish(raw: &JsonValue, fields: &[&str]) -> Option<String> {
2350 fields.iter().find_map(|field| {
2351 let value = raw.get(*field)?;
2352 value
2353 .as_str()
2354 .map(ToString::to_string)
2355 .or_else(|| parse_json_i64ish(value).map(|number| number.to_string()))
2356 .or_else(|| value.as_u64().map(|number| number.to_string()))
2357 })
2358}
2359
2360fn parse_rfc3339(text: &str) -> Option<OffsetDateTime> {
2361 OffsetDateTime::parse(text, &time::format_description::well_known::Rfc3339).ok()
2362}
2363
2364#[cfg(test)]
2365mod tests {
2366 use super::*;
2367
2368 fn sample_headers() -> BTreeMap<String, String> {
2369 BTreeMap::from([
2370 ("Authorization".to_string(), "Bearer secret".to_string()),
2371 ("Cookie".to_string(), "session=abc".to_string()),
2372 ("User-Agent".to_string(), "GitHub-Hookshot/123".to_string()),
2373 ("X-GitHub-Delivery".to_string(), "delivery-123".to_string()),
2374 ("X-GitHub-Event".to_string(), "issues".to_string()),
2375 ("X-Webhook-Token".to_string(), "token".to_string()),
2376 ])
2377 }
2378
2379 #[test]
2380 fn default_redaction_policy_keeps_safe_headers() {
2381 let redacted = redact_headers(&sample_headers(), &HeaderRedactionPolicy::default());
2382 assert_eq!(redacted.get("User-Agent").unwrap(), "GitHub-Hookshot/123");
2383 assert_eq!(redacted.get("X-GitHub-Delivery").unwrap(), "delivery-123");
2384 assert_eq!(
2385 redacted.get("Authorization").unwrap(),
2386 REDACTED_HEADER_VALUE
2387 );
2388 assert_eq!(redacted.get("Cookie").unwrap(), REDACTED_HEADER_VALUE);
2389 assert_eq!(
2390 redacted.get("X-Webhook-Token").unwrap(),
2391 REDACTED_HEADER_VALUE
2392 );
2393 }
2394
2395 #[test]
2396 fn provider_catalog_rejects_duplicates() {
2397 let mut catalog = ProviderCatalog::default();
2398 catalog
2399 .register(Arc::new(BuiltinProviderSchema {
2400 provider_id: "github",
2401 harn_schema_name: "GitHubEventPayload",
2402 metadata: provider_metadata_entry(
2403 "github",
2404 &["webhook"],
2405 "GitHubEventPayload",
2406 &[],
2407 SignatureVerificationMetadata::None,
2408 Vec::new(),
2409 ProviderRuntimeMetadata::Placeholder,
2410 ),
2411 normalize: github_payload,
2412 }))
2413 .unwrap();
2414 let error = catalog
2415 .register(Arc::new(BuiltinProviderSchema {
2416 provider_id: "github",
2417 harn_schema_name: "GitHubEventPayload",
2418 metadata: provider_metadata_entry(
2419 "github",
2420 &["webhook"],
2421 "GitHubEventPayload",
2422 &[],
2423 SignatureVerificationMetadata::None,
2424 Vec::new(),
2425 ProviderRuntimeMetadata::Placeholder,
2426 ),
2427 normalize: github_payload,
2428 }))
2429 .unwrap_err();
2430 assert_eq!(
2431 error,
2432 ProviderCatalogError::DuplicateProvider("github".to_string())
2433 );
2434 }
2435
2436 #[test]
2437 fn registered_provider_metadata_marks_builtin_connectors() {
2438 let entries = registered_provider_metadata();
2439 let builtin: Vec<&ProviderMetadata> = entries
2440 .iter()
2441 .filter(|entry| matches!(entry.runtime, ProviderRuntimeMetadata::Builtin { .. }))
2442 .collect();
2443
2444 assert_eq!(builtin.len(), 9);
2445 assert!(builtin.iter().any(|entry| entry.provider == "a2a-push"));
2446 assert!(builtin.iter().any(|entry| entry.provider == "cron"));
2447 assert!(builtin.iter().any(|entry| entry.provider == "webhook"));
2448 for provider in ["github", "linear", "notion", "slack"] {
2449 let entry = entries
2450 .iter()
2451 .find(|entry| entry.provider == provider)
2452 .expect("first-party package-backed provider metadata");
2453 assert!(matches!(
2454 entry.runtime,
2455 ProviderRuntimeMetadata::Placeholder
2456 ));
2457 }
2458 let kafka = entries
2459 .iter()
2460 .find(|entry| entry.provider == "kafka")
2461 .expect("kafka stream provider");
2462 assert_eq!(kafka.kinds, vec!["stream".to_string()]);
2463 assert_eq!(kafka.schema_name, "StreamEventPayload");
2464 assert!(matches!(
2465 kafka.runtime,
2466 ProviderRuntimeMetadata::Builtin {
2467 ref connector,
2468 default_signature_variant: None
2469 } if connector == "stream"
2470 ));
2471 }
2472
2473 #[test]
2474 fn trigger_event_round_trip_is_stable() {
2475 let provider = ProviderId::from("github");
2476 let headers = redact_headers(&sample_headers(), &HeaderRedactionPolicy::default());
2477 let payload = ProviderPayload::normalize(
2478 &provider,
2479 "issues",
2480 &sample_headers(),
2481 serde_json::json!({
2482 "action": "opened",
2483 "installation": {"id": 42},
2484 "issue": {"number": 99}
2485 }),
2486 )
2487 .unwrap();
2488 let event = TriggerEvent {
2489 id: TriggerEventId("trigger_evt_fixed".to_string()),
2490 provider,
2491 kind: "issues".to_string(),
2492 received_at: parse_rfc3339("2026-04-19T07:00:00Z").unwrap(),
2493 occurred_at: Some(parse_rfc3339("2026-04-19T06:59:59Z").unwrap()),
2494 dedupe_key: "delivery-123".to_string(),
2495 trace_id: TraceId("trace_fixed".to_string()),
2496 tenant_id: Some(TenantId("tenant_1".to_string())),
2497 headers,
2498 provider_payload: payload,
2499 signature_status: SignatureStatus::Verified,
2500 dedupe_claimed: false,
2501 batch: None,
2502 raw_body: Some(vec![0, 159, 255, 10]),
2503 };
2504
2505 let once = serde_json::to_value(&event).unwrap();
2506 assert_eq!(once["raw_body"], serde_json::json!("AJ//Cg=="));
2507 let decoded: TriggerEvent = serde_json::from_value(once.clone()).unwrap();
2508 let twice = serde_json::to_value(&decoded).unwrap();
2509 assert_eq!(decoded, event);
2510 assert_eq!(once, twice);
2511 }
2512
2513 #[test]
2514 fn unknown_provider_errors() {
2515 let error = ProviderPayload::normalize(
2516 &ProviderId::from("custom-provider"),
2517 "thing.happened",
2518 &BTreeMap::new(),
2519 serde_json::json!({"ok": true}),
2520 )
2521 .unwrap_err();
2522 assert_eq!(
2523 error,
2524 ProviderCatalogError::UnknownProvider("custom-provider".to_string())
2525 );
2526 }
2527
2528 fn github_headers(event: &str, delivery: &str) -> BTreeMap<String, String> {
2529 BTreeMap::from([
2530 ("X-GitHub-Event".to_string(), event.to_string()),
2531 ("X-GitHub-Delivery".to_string(), delivery.to_string()),
2532 ])
2533 }
2534
2535 fn unwrap_github(payload: ProviderPayload) -> GitHubEventPayload {
2536 match payload {
2537 ProviderPayload::Known(KnownProviderPayload::GitHub(p)) => p,
2538 other => panic!("expected GitHub payload, got {other:?}"),
2539 }
2540 }
2541
2542 fn connector_normalized(
2546 event: &str,
2547 delivery: &str,
2548 installation_id: i64,
2549 action: Option<&str>,
2550 original: serde_json::Value,
2551 promoted: serde_json::Value,
2552 ) -> serde_json::Value {
2553 let mut common = serde_json::json!({
2554 "provider": "github",
2555 "event": event,
2556 "topic": match action {
2557 Some(a) => format!("github.{event}.{a}"),
2558 None => format!("github.{event}"),
2559 },
2560 "delivery_id": delivery,
2561 "installation_id": installation_id,
2562 "repository": original.get("repository").cloned().unwrap_or(JsonValue::Null),
2563 "repo": serde_json::json!({"owner": "octo-org", "name": "octo-repo", "full_name": "octo-org/octo-repo"}),
2564 "raw": original,
2565 });
2566 if let Some(a) = action {
2567 common["action"] = serde_json::json!(a);
2568 }
2569 let common_obj = common.as_object_mut().unwrap();
2570 if let Some(promoted_obj) = promoted.as_object() {
2571 for (k, v) in promoted_obj {
2572 common_obj.insert(k.clone(), v.clone());
2573 }
2574 }
2575 common
2576 }
2577
2578 #[test]
2579 fn github_check_suite_event_promotes_typed_fields() {
2580 let original = serde_json::json!({
2581 "action": "requested",
2582 "check_suite": {
2583 "id": 8101,
2584 "status": "queued",
2585 "conclusion": null,
2586 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2587 "head_branch": "feature/x",
2588 },
2589 "repository": {"full_name": "octo-org/octo-repo"},
2590 "installation": {"id": 3001},
2591 });
2592 let normalized = connector_normalized(
2593 "check_suite",
2594 "delivery-cs",
2595 3001,
2596 Some("requested"),
2597 original.clone(),
2598 serde_json::json!({
2599 "check_suite": original["check_suite"].clone(),
2600 "check_suite_id": 8101,
2601 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2602 "head_ref": "feature/x",
2603 "status": "queued",
2604 }),
2605 );
2606 let provider = ProviderId::from("github");
2607 let payload = ProviderPayload::normalize(
2608 &provider,
2609 "check_suite",
2610 &github_headers("check_suite", "delivery-cs"),
2611 normalized,
2612 )
2613 .expect("check_suite payload");
2614 let GitHubEventPayload::CheckSuite(check_suite) = unwrap_github(payload) else {
2615 panic!("expected CheckSuite variant");
2616 };
2617 assert_eq!(check_suite.common.event, "check_suite");
2618 assert_eq!(check_suite.common.action.as_deref(), Some("requested"));
2619 assert_eq!(
2620 check_suite.common.delivery_id.as_deref(),
2621 Some("delivery-cs")
2622 );
2623 assert_eq!(check_suite.common.installation_id, Some(3001));
2624 assert_eq!(
2625 check_suite.common.topic.as_deref(),
2626 Some("github.check_suite.requested")
2627 );
2628 assert!(check_suite.common.repository.is_some());
2629 assert!(check_suite.common.repo.is_some());
2630 assert_eq!(check_suite.check_suite_id, Some(8101));
2631 assert_eq!(
2632 check_suite.head_sha.as_deref(),
2633 Some("ccccccccccccccccccccccccccccccccccccccc1")
2634 );
2635 assert_eq!(check_suite.head_ref.as_deref(), Some("feature/x"));
2636 assert_eq!(check_suite.status.as_deref(), Some("queued"));
2637 assert_eq!(check_suite.common.raw, original);
2639 }
2640
2641 #[test]
2642 fn github_status_event_promotes_typed_fields() {
2643 let original = serde_json::json!({
2644 "id": 9101,
2645 "sha": "ccccccccccccccccccccccccccccccccccccccc1",
2646 "state": "success",
2647 "context": "legacy/status",
2648 "target_url": "https://ci.example.test/octo-repo/9101",
2649 "branches": [{"name": "main"}],
2650 "repository": {"full_name": "octo-org/octo-repo"},
2651 "installation": {"id": 3001},
2652 });
2653 let normalized = connector_normalized(
2654 "status",
2655 "delivery-status",
2656 3001,
2657 None,
2658 original.clone(),
2659 serde_json::json!({
2660 "commit_status": original.clone(),
2661 "status_id": 9101,
2662 "head_sha": "ccccccccccccccccccccccccccccccccccccccc1",
2663 "head_ref": "main",
2664 "base_ref": "main",
2665 "state": "success",
2666 "context": "legacy/status",
2667 "target_url": "https://ci.example.test/octo-repo/9101",
2668 }),
2669 );
2670 let provider = ProviderId::from("github");
2671 let payload = ProviderPayload::normalize(
2672 &provider,
2673 "status",
2674 &github_headers("status", "delivery-status"),
2675 normalized,
2676 )
2677 .expect("status payload");
2678 let GitHubEventPayload::Status(status) = unwrap_github(payload) else {
2679 panic!("expected Status variant");
2680 };
2681 assert_eq!(status.common.event, "status");
2682 assert_eq!(status.common.installation_id, Some(3001));
2683 assert_eq!(status.status_id, Some(9101));
2684 assert_eq!(status.state.as_deref(), Some("success"));
2685 assert_eq!(status.context.as_deref(), Some("legacy/status"));
2686 assert_eq!(
2687 status.target_url.as_deref(),
2688 Some("https://ci.example.test/octo-repo/9101")
2689 );
2690 assert_eq!(
2691 status.head_sha.as_deref(),
2692 Some("ccccccccccccccccccccccccccccccccccccccc1")
2693 );
2694 assert!(status.commit_status.is_some());
2695 }
2696
2697 #[test]
2698 fn github_merge_group_event_promotes_typed_fields() {
2699 let original = serde_json::json!({
2700 "action": "checks_requested",
2701 "merge_group": {
2702 "id": 9201,
2703 "head_ref": "gh-readonly-queue/main/pr-42",
2704 "head_sha": "ddddddddddddddddddddddddddddddddddddddd1",
2705 "base_ref": "main",
2706 "base_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
2707 "pull_requests": [{"number": 42}, {"number": 43}],
2708 },
2709 "repository": {"full_name": "octo-org/octo-repo"},
2710 "installation": {"id": 3001},
2711 });
2712 let normalized = connector_normalized(
2713 "merge_group",
2714 "delivery-mg",
2715 3001,
2716 Some("checks_requested"),
2717 original.clone(),
2718 serde_json::json!({
2719 "merge_group": original["merge_group"].clone(),
2720 "merge_group_id": 9201,
2721 "head_sha": "ddddddddddddddddddddddddddddddddddddddd1",
2722 "head_ref": "gh-readonly-queue/main/pr-42",
2723 "base_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
2724 "base_ref": "main",
2725 "pull_requests": [{"number": 42}, {"number": 43}],
2726 "pull_request_numbers": [42, 43],
2727 }),
2728 );
2729 let provider = ProviderId::from("github");
2730 let payload = ProviderPayload::normalize(
2731 &provider,
2732 "merge_group",
2733 &github_headers("merge_group", "delivery-mg"),
2734 normalized,
2735 )
2736 .expect("merge_group payload");
2737 let GitHubEventPayload::MergeGroup(mg) = unwrap_github(payload) else {
2738 panic!("expected MergeGroup variant");
2739 };
2740 assert_eq!(mg.common.event, "merge_group");
2741 assert_eq!(mg.common.action.as_deref(), Some("checks_requested"));
2742 assert_eq!(mg.merge_group_id, Some(serde_json::json!(9201)));
2743 assert_eq!(mg.head_ref.as_deref(), Some("gh-readonly-queue/main/pr-42"));
2744 assert_eq!(
2745 mg.base_sha.as_deref(),
2746 Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1")
2747 );
2748 assert_eq!(mg.base_ref.as_deref(), Some("main"));
2749 assert_eq!(mg.pull_request_numbers, vec![42i64, 43i64]);
2750 assert_eq!(mg.pull_requests.len(), 2);
2751 }
2752
2753 #[test]
2754 fn github_installation_event_promotes_typed_fields() {
2755 let original = serde_json::json!({
2756 "action": "suspend",
2757 "installation": {
2758 "id": 3001,
2759 "account": {"login": "octo-org"},
2760 "repository_selection": "selected",
2761 "suspended_at": "2026-04-20T18:00:00Z",
2762 },
2763 "repositories": [{"full_name": "octo-org/octo-repo"}],
2764 });
2765 let normalized = connector_normalized(
2766 "installation",
2767 "delivery-inst",
2768 3001,
2769 Some("suspend"),
2770 original.clone(),
2771 serde_json::json!({
2772 "installation": original["installation"].clone(),
2773 "account": {"login": "octo-org"},
2774 "installation_state": "suspended",
2775 "suspended": true,
2776 "revoked": false,
2777 "repositories": original["repositories"].clone(),
2778 }),
2779 );
2780 let provider = ProviderId::from("github");
2781 let payload = ProviderPayload::normalize(
2782 &provider,
2783 "installation",
2784 &github_headers("installation", "delivery-inst"),
2785 normalized,
2786 )
2787 .expect("installation payload");
2788 let GitHubEventPayload::Installation(inst) = unwrap_github(payload) else {
2789 panic!("expected Installation variant");
2790 };
2791 assert_eq!(inst.common.event, "installation");
2792 assert_eq!(inst.common.action.as_deref(), Some("suspend"));
2793 assert_eq!(inst.installation_state.as_deref(), Some("suspended"));
2794 assert_eq!(inst.suspended, Some(true));
2795 assert_eq!(inst.revoked, Some(false));
2796 assert_eq!(inst.repositories.len(), 1);
2797 assert!(inst.account.is_some());
2798 }
2799
2800 #[test]
2801 fn github_installation_repositories_event_promotes_typed_fields() {
2802 let original = serde_json::json!({
2803 "action": "removed",
2804 "installation": {"id": 3001, "account": {"login": "octo-org"}},
2805 "repository_selection": "selected",
2806 "repositories_added": [],
2807 "repositories_removed": [
2808 {"id": 4001, "full_name": "octo-org/octo-repo"},
2809 ],
2810 });
2811 let normalized = connector_normalized(
2812 "installation_repositories",
2813 "delivery-inst-repos",
2814 3001,
2815 Some("removed"),
2816 original.clone(),
2817 serde_json::json!({
2818 "installation": original["installation"].clone(),
2819 "account": {"login": "octo-org"},
2820 "installation_state": "revoked",
2821 "suspended": false,
2822 "revoked": true,
2823 "repository_selection": "selected",
2824 "repositories_added": [],
2825 "repositories_removed": original["repositories_removed"].clone(),
2826 }),
2827 );
2828 let provider = ProviderId::from("github");
2829 let payload = ProviderPayload::normalize(
2830 &provider,
2831 "installation_repositories",
2832 &github_headers("installation_repositories", "delivery-inst-repos"),
2833 normalized,
2834 )
2835 .expect("installation_repositories payload");
2836 let GitHubEventPayload::InstallationRepositories(repos) = unwrap_github(payload) else {
2837 panic!("expected InstallationRepositories variant");
2838 };
2839 assert_eq!(repos.common.event, "installation_repositories");
2840 assert_eq!(repos.common.action.as_deref(), Some("removed"));
2841 assert_eq!(repos.repository_selection.as_deref(), Some("selected"));
2842 assert!(repos.repositories_added.is_empty());
2843 assert_eq!(repos.repositories_removed.len(), 1);
2844 assert_eq!(
2845 repos.repositories_removed[0]
2846 .get("full_name")
2847 .and_then(|v| v.as_str()),
2848 Some("octo-org/octo-repo"),
2849 );
2850 assert_eq!(repos.installation_state.as_deref(), Some("revoked"));
2851 assert_eq!(repos.revoked, Some(true));
2852 }
2853
2854 #[test]
2855 fn github_legacy_direct_webhook_still_normalizes() {
2856 let provider = ProviderId::from("github");
2860 let payload = ProviderPayload::normalize(
2861 &provider,
2862 "issues",
2863 &github_headers("issues", "delivery-legacy"),
2864 serde_json::json!({
2865 "action": "opened",
2866 "installation": {"id": 99},
2867 "issue": {"number": 7},
2868 }),
2869 )
2870 .expect("legacy issues payload");
2871 let GitHubEventPayload::Issues(issues) = unwrap_github(payload) else {
2872 panic!("expected Issues variant");
2873 };
2874 assert_eq!(issues.common.installation_id, Some(99));
2875 assert_eq!(
2876 issues.common.delivery_id.as_deref(),
2877 Some("delivery-legacy")
2878 );
2879 assert!(issues.common.topic.is_none());
2880 assert!(issues.common.repo.is_none());
2881 assert_eq!(issues.issue.get("number").and_then(|v| v.as_i64()), Some(7));
2882 }
2883
2884 #[test]
2885 fn github_new_event_variants_round_trip_through_serde() {
2886 let provider = ProviderId::from("github");
2891 let cases: &[(&str, serde_json::Value, &str)] = &[
2892 (
2893 "check_suite",
2894 serde_json::json!({
2895 "event": "check_suite",
2896 "check_suite": {"id": 1},
2897 "check_suite_id": 1,
2898 "raw": {"check_suite": {"id": 1}},
2899 }),
2900 "CheckSuite",
2901 ),
2902 (
2903 "status",
2904 serde_json::json!({
2905 "event": "status",
2906 "commit_status": {"id": 9},
2907 "status_id": 9,
2908 "state": "success",
2909 "raw": {"id": 9, "state": "success"},
2910 }),
2911 "Status",
2912 ),
2913 (
2914 "merge_group",
2915 serde_json::json!({
2916 "event": "merge_group",
2917 "merge_group": {"id": 1},
2918 "merge_group_id": 1,
2919 "raw": {"merge_group": {"id": 1}},
2920 }),
2921 "MergeGroup",
2922 ),
2923 (
2924 "installation",
2925 serde_json::json!({
2926 "event": "installation",
2927 "installation": {"id": 1},
2928 "installation_state": "active",
2929 "suspended": false,
2930 "raw": {"installation": {"id": 1}},
2931 }),
2932 "Installation",
2933 ),
2934 (
2935 "installation_repositories",
2936 serde_json::json!({
2937 "event": "installation_repositories",
2938 "installation": {"id": 1},
2939 "repository_selection": "selected",
2940 "repositories_added": [],
2941 "repositories_removed": [{"id": 7}],
2942 "raw": {"installation": {"id": 1}},
2943 }),
2944 "InstallationRepositories",
2945 ),
2946 ];
2947 for (kind, raw, want_variant) in cases {
2948 let payload = ProviderPayload::normalize(
2949 &provider,
2950 kind,
2951 &github_headers(kind, "delivery"),
2952 raw.clone(),
2953 )
2954 .unwrap_or_else(|_| panic!("normalize {kind}"));
2955 let serialized = serde_json::to_value(&payload).expect("serialize");
2956 let deserialized: ProviderPayload =
2957 serde_json::from_value(serialized.clone()).expect("deserialize");
2958 let actual_variant = match unwrap_github(deserialized) {
2959 GitHubEventPayload::Issues(_) => "Issues",
2960 GitHubEventPayload::PullRequest(_) => "PullRequest",
2961 GitHubEventPayload::IssueComment(_) => "IssueComment",
2962 GitHubEventPayload::PullRequestReview(_) => "PullRequestReview",
2963 GitHubEventPayload::Push(_) => "Push",
2964 GitHubEventPayload::WorkflowRun(_) => "WorkflowRun",
2965 GitHubEventPayload::DeploymentStatus(_) => "DeploymentStatus",
2966 GitHubEventPayload::CheckRun(_) => "CheckRun",
2967 GitHubEventPayload::CheckSuite(_) => "CheckSuite",
2968 GitHubEventPayload::Status(_) => "Status",
2969 GitHubEventPayload::MergeGroup(_) => "MergeGroup",
2970 GitHubEventPayload::Installation(_) => "Installation",
2971 GitHubEventPayload::InstallationRepositories(_) => "InstallationRepositories",
2972 GitHubEventPayload::Other(_) => "Other",
2973 };
2974 assert_eq!(
2975 actual_variant, *want_variant,
2976 "{kind} round-tripped as {actual_variant}, expected {want_variant}; serialized form: {serialized}"
2977 );
2978 }
2979 }
2980
2981 #[test]
2982 fn provider_normalizes_stream_payloads() {
2983 let payload = ProviderPayload::normalize(
2984 &ProviderId::from("kafka"),
2985 "quote.tick",
2986 &BTreeMap::from([("x-source".to_string(), "feed".to_string())]),
2987 serde_json::json!({
2988 "topic": "quotes",
2989 "partition": 7,
2990 "offset": "42",
2991 "key": "AAPL",
2992 "timestamp": "2026-04-21T12:00:00Z"
2993 }),
2994 )
2995 .expect("stream payload");
2996 let ProviderPayload::Known(KnownProviderPayload::Kafka(payload)) = payload else {
2997 panic!("expected kafka stream payload")
2998 };
2999 assert_eq!(payload.event, "quote.tick");
3000 assert_eq!(payload.stream.as_deref(), Some("quotes"));
3001 assert_eq!(payload.partition.as_deref(), Some("7"));
3002 assert_eq!(payload.offset.as_deref(), Some("42"));
3003 assert_eq!(payload.key.as_deref(), Some("AAPL"));
3004 assert_eq!(payload.timestamp.as_deref(), Some("2026-04-21T12:00:00Z"));
3005 }
3006}