1use std::collections::{BTreeMap, BTreeSet};
2use std::sync::{Arc, OnceLock, RwLock};
3
4use serde::{Deserialize, Serialize};
5use serde_json::Value as JsonValue;
6
7use super::core::ProviderId;
8use super::normalize::{
9 a2a_push_payload, cron_payload, email_payload, github_payload, kafka_payload, linear_payload,
10 nats_payload, notion_payload, postgres_cdc_payload, pulsar_payload, slack_payload,
11 webhook_payload, websocket_payload,
12};
13use super::payloads::ProviderPayload;
14
15impl ProviderPayload {
16 pub fn normalize(
17 provider: &ProviderId,
18 kind: &str,
19 headers: &BTreeMap<String, String>,
20 raw: JsonValue,
21 ) -> Result<Self, ProviderCatalogError> {
22 provider_catalog()
23 .read()
24 .expect("provider catalog poisoned")
25 .normalize(provider, kind, headers, raw)
26 }
27}
28
29#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
30pub struct ProviderSecretRequirement {
31 pub name: String,
32 pub required: bool,
33 pub namespace: String,
34}
35
36#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
37pub struct ProviderOutboundMethod {
38 pub name: String,
39}
40
41#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
42#[serde(tag = "kind", rename_all = "snake_case")]
43pub enum SignatureVerificationMetadata {
44 #[default]
45 None,
46 Hmac {
47 variant: String,
48 raw_body: bool,
49 signature_header: String,
50 timestamp_header: Option<String>,
51 id_header: Option<String>,
52 default_tolerance_secs: Option<i64>,
53 digest: String,
54 encoding: String,
55 },
56}
57
58#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
59#[serde(tag = "kind", rename_all = "snake_case")]
60pub enum ProviderRuntimeMetadata {
61 Builtin {
62 connector: String,
63 default_signature_variant: Option<String>,
64 },
65 #[default]
66 Placeholder,
67}
68
69#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
70pub struct ProviderMetadata {
71 pub provider: String,
72 #[serde(default)]
73 pub kinds: Vec<String>,
74 pub schema_name: String,
75 #[serde(default)]
76 pub outbound_methods: Vec<ProviderOutboundMethod>,
77 #[serde(default)]
78 pub secret_requirements: Vec<ProviderSecretRequirement>,
79 #[serde(default)]
80 pub signature_verification: SignatureVerificationMetadata,
81 #[serde(default)]
82 pub runtime: ProviderRuntimeMetadata,
83}
84
85impl ProviderMetadata {
86 pub fn supports_kind(&self, kind: &str) -> bool {
87 self.kinds.iter().any(|candidate| candidate == kind)
88 }
89
90 pub fn required_secret_names(&self) -> impl Iterator<Item = &str> {
91 self.secret_requirements
92 .iter()
93 .filter(|requirement| requirement.required)
94 .map(|requirement| requirement.name.as_str())
95 }
96}
97
98pub trait ProviderSchema: Send + Sync {
99 fn provider_id(&self) -> &str;
100 fn harn_schema_name(&self) -> &str;
101 fn metadata(&self) -> ProviderMetadata {
102 ProviderMetadata {
103 provider: self.provider_id().to_string(),
104 schema_name: self.harn_schema_name().to_string(),
105 ..ProviderMetadata::default()
106 }
107 }
108 fn normalize(
109 &self,
110 kind: &str,
111 headers: &BTreeMap<String, String>,
112 raw: JsonValue,
113 ) -> Result<ProviderPayload, ProviderCatalogError>;
114}
115
116#[derive(Clone, Debug, PartialEq, Eq)]
117pub enum ProviderCatalogError {
118 DuplicateProvider(String),
119 UnknownProvider(String),
120}
121
122impl std::fmt::Display for ProviderCatalogError {
123 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124 match self {
125 Self::DuplicateProvider(provider) => {
126 write!(f, "provider `{provider}` is already registered")
127 }
128 Self::UnknownProvider(provider) => write!(f, "provider `{provider}` is not registered"),
129 }
130 }
131}
132
133impl std::error::Error for ProviderCatalogError {}
134
135#[derive(Clone, Default)]
136pub struct ProviderCatalog {
137 providers: BTreeMap<String, Arc<dyn ProviderSchema>>,
138}
139
140impl ProviderCatalog {
141 pub fn with_defaults() -> Self {
142 let mut catalog = Self::default();
143 for schema in default_provider_schemas() {
144 catalog
145 .register(schema)
146 .expect("default providers must register cleanly");
147 }
148 catalog
149 }
150
151 pub fn with_defaults_and(
152 schemas: Vec<Arc<dyn ProviderSchema>>,
153 ) -> Result<Self, ProviderCatalogError> {
154 let mut catalog = Self::with_defaults();
155 let builtin_providers: BTreeSet<String> = catalog.schema_names().into_keys().collect();
156 for schema in schemas {
157 if builtin_providers.contains(schema.provider_id()) {
158 continue;
159 }
160 catalog.register(schema)?;
161 }
162 Ok(catalog)
163 }
164
165 pub fn register(
166 &mut self,
167 schema: Arc<dyn ProviderSchema>,
168 ) -> Result<(), ProviderCatalogError> {
169 let provider = schema.provider_id().to_string();
170 if self.providers.contains_key(provider.as_str()) {
171 return Err(ProviderCatalogError::DuplicateProvider(provider));
172 }
173 self.providers.insert(provider, schema);
174 Ok(())
175 }
176
177 pub fn normalize(
178 &self,
179 provider: &ProviderId,
180 kind: &str,
181 headers: &BTreeMap<String, String>,
182 raw: JsonValue,
183 ) -> Result<ProviderPayload, ProviderCatalogError> {
184 let schema = self
185 .providers
186 .get(provider.as_str())
187 .ok_or_else(|| ProviderCatalogError::UnknownProvider(provider.0.clone()))?;
188 schema.normalize(kind, headers, raw)
189 }
190
191 pub fn schema_names(&self) -> BTreeMap<String, String> {
192 self.providers
193 .iter()
194 .map(|(provider, schema)| (provider.clone(), schema.harn_schema_name().to_string()))
195 .collect()
196 }
197
198 pub fn entries(&self) -> Vec<ProviderMetadata> {
199 self.providers
200 .values()
201 .map(|schema| schema.metadata())
202 .collect()
203 }
204
205 pub fn metadata_for(&self, provider: &str) -> Option<ProviderMetadata> {
206 self.providers.get(provider).map(|schema| schema.metadata())
207 }
208}
209
210pub fn register_provider_schema(
211 schema: Arc<dyn ProviderSchema>,
212) -> Result<(), ProviderCatalogError> {
213 provider_catalog()
214 .write()
215 .expect("provider catalog poisoned")
216 .register(schema)
217}
218
219pub fn reset_provider_catalog() {
220 *provider_catalog()
221 .write()
222 .expect("provider catalog poisoned") = ProviderCatalog::with_defaults();
223}
224
225pub fn reset_provider_catalog_with(
226 schemas: Vec<Arc<dyn ProviderSchema>>,
227) -> Result<(), ProviderCatalogError> {
228 let catalog = ProviderCatalog::with_defaults_and(schemas)?;
229 install_provider_catalog(catalog);
230 Ok(())
231}
232
233pub fn install_provider_catalog(catalog: ProviderCatalog) {
234 *provider_catalog()
235 .write()
236 .expect("provider catalog poisoned") = catalog;
237}
238
239pub fn registered_provider_schema_names() -> BTreeMap<String, String> {
240 provider_catalog()
241 .read()
242 .expect("provider catalog poisoned")
243 .schema_names()
244}
245
246pub fn registered_provider_metadata() -> Vec<ProviderMetadata> {
247 provider_catalog()
248 .read()
249 .expect("provider catalog poisoned")
250 .entries()
251}
252
253pub fn provider_metadata(provider: &str) -> Option<ProviderMetadata> {
254 provider_catalog()
255 .read()
256 .expect("provider catalog poisoned")
257 .metadata_for(provider)
258}
259
260fn provider_catalog() -> &'static RwLock<ProviderCatalog> {
261 static PROVIDER_CATALOG: OnceLock<RwLock<ProviderCatalog>> = OnceLock::new();
262 PROVIDER_CATALOG.get_or_init(|| RwLock::new(ProviderCatalog::with_defaults()))
263}
264
265struct BuiltinProviderSchema {
266 provider_id: &'static str,
267 harn_schema_name: &'static str,
268 metadata: ProviderMetadata,
269 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
270}
271
272impl ProviderSchema for BuiltinProviderSchema {
273 fn provider_id(&self) -> &str {
274 self.provider_id
275 }
276
277 fn harn_schema_name(&self) -> &str {
278 self.harn_schema_name
279 }
280
281 fn metadata(&self) -> ProviderMetadata {
282 self.metadata.clone()
283 }
284
285 fn normalize(
286 &self,
287 kind: &str,
288 headers: &BTreeMap<String, String>,
289 raw: JsonValue,
290 ) -> Result<ProviderPayload, ProviderCatalogError> {
291 Ok((self.normalize)(kind, headers, raw))
292 }
293}
294
295fn provider_metadata_entry(
296 provider: &str,
297 kinds: &[&str],
298 schema_name: &str,
299 outbound_methods: &[&str],
300 signature_verification: SignatureVerificationMetadata,
301 secret_requirements: Vec<ProviderSecretRequirement>,
302 runtime: ProviderRuntimeMetadata,
303) -> ProviderMetadata {
304 ProviderMetadata {
305 provider: provider.to_string(),
306 kinds: kinds.iter().map(|kind| kind.to_string()).collect(),
307 schema_name: schema_name.to_string(),
308 outbound_methods: outbound_methods
309 .iter()
310 .map(|name| ProviderOutboundMethod {
311 name: (*name).to_string(),
312 })
313 .collect(),
314 secret_requirements,
315 signature_verification,
316 runtime,
317 }
318}
319
320fn hmac_signature_metadata(
321 variant: &str,
322 signature_header: &str,
323 timestamp_header: Option<&str>,
324 id_header: Option<&str>,
325 default_tolerance_secs: Option<i64>,
326 encoding: &str,
327) -> SignatureVerificationMetadata {
328 SignatureVerificationMetadata::Hmac {
329 variant: variant.to_string(),
330 raw_body: true,
331 signature_header: signature_header.to_string(),
332 timestamp_header: timestamp_header.map(ToString::to_string),
333 id_header: id_header.map(ToString::to_string),
334 default_tolerance_secs,
335 digest: "sha256".to_string(),
336 encoding: encoding.to_string(),
337 }
338}
339
340fn required_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
341 ProviderSecretRequirement {
342 name: name.to_string(),
343 required: true,
344 namespace: namespace.to_string(),
345 }
346}
347
348fn outbound_method(name: &str) -> ProviderOutboundMethod {
349 ProviderOutboundMethod {
350 name: name.to_string(),
351 }
352}
353
354fn optional_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
355 ProviderSecretRequirement {
356 name: name.to_string(),
357 required: false,
358 namespace: namespace.to_string(),
359 }
360}
361
362fn default_provider_schemas() -> Vec<Arc<dyn ProviderSchema>> {
363 vec![
364 Arc::new(BuiltinProviderSchema {
365 provider_id: "github",
366 harn_schema_name: "GitHubEventPayload",
367 metadata: provider_metadata_entry(
368 "github",
369 &["webhook"],
370 "GitHubEventPayload",
371 &[
372 "github.pr.list",
373 "github.pr.view",
374 "github.pr.edit",
375 "github.pr.checks",
376 "github.pr.merge",
377 "github.pr.enable_auto_merge",
378 "github.pr.comment",
379 "github.actions.workflow_dispatch",
380 "github.actions.runs",
381 "github.actions.run",
382 "github.actions.logs",
383 "github.release.view",
384 "github.release.latest",
385 "github.release.assets",
386 "github.merge_queue.entries",
387 "github.merge_queue.enqueue",
388 "github.issue.create",
389 "github.issue.comment",
390 "github.branch.protection",
391 "api_call",
392 "issues.create_comment",
393 "issues.create",
394 "issues.create_with_template",
395 "issues.update",
396 "issues.add_labels",
397 "pulls.list",
398 "pulls.list_with_checks",
399 "pulls.get",
400 "pulls.update",
401 "pulls.merge",
402 "pulls.merge_safe",
403 "pulls.create_review_comment",
404 "pulls.get_diff",
405 "pulls.list_files",
406 "pulls.list_reviews",
407 "repos.get_content",
408 "repos.get_text",
409 "repos.get_latest_release",
410 "repos.list_release_assets",
411 "repos.get_branch_protection",
412 "git.delete_ref",
413 "actions.workflow_dispatch",
414 "actions.workflow_runs.list",
415 "actions.workflow_run.get",
416 "check_runs.create",
417 "check_runs.update",
418 "graphql",
419 ],
420 hmac_signature_metadata(
421 "github",
422 "X-Hub-Signature-256",
423 None,
424 Some("X-GitHub-Delivery"),
425 None,
426 "hex",
427 ),
428 vec![required_secret("signing_secret", "github")],
429 ProviderRuntimeMetadata::Placeholder,
430 ),
431 normalize: github_payload,
432 }),
433 Arc::new(BuiltinProviderSchema {
434 provider_id: "slack",
435 harn_schema_name: "SlackEventPayload",
436 metadata: provider_metadata_entry(
437 "slack",
438 &["webhook"],
439 "SlackEventPayload",
440 &[
441 "post_message",
442 "update_message",
443 "add_reaction",
444 "open_view",
445 "user_info",
446 "api_call",
447 "upload_file",
448 ],
449 hmac_signature_metadata(
450 "slack",
451 "X-Slack-Signature",
452 Some("X-Slack-Request-Timestamp"),
453 None,
454 Some(300),
455 "hex",
456 ),
457 vec![required_secret("signing_secret", "slack")],
458 ProviderRuntimeMetadata::Placeholder,
459 ),
460 normalize: slack_payload,
461 }),
462 Arc::new(BuiltinProviderSchema {
463 provider_id: "linear",
464 harn_schema_name: "LinearEventPayload",
465 metadata: {
466 let mut metadata = provider_metadata_entry(
467 "linear",
468 &["webhook"],
469 "LinearEventPayload",
470 &[],
471 hmac_signature_metadata(
472 "linear",
473 "Linear-Signature",
474 None,
475 Some("Linear-Delivery"),
476 Some(75),
477 "hex",
478 ),
479 vec![
480 required_secret("signing_secret", "linear"),
481 optional_secret("access_token", "linear"),
482 ],
483 ProviderRuntimeMetadata::Placeholder,
484 );
485 metadata.outbound_methods = vec![
486 outbound_method("list_issues"),
487 outbound_method("update_issue"),
488 outbound_method("create_comment"),
489 outbound_method("search"),
490 outbound_method("graphql"),
491 ];
492 metadata
493 },
494 normalize: linear_payload,
495 }),
496 Arc::new(BuiltinProviderSchema {
497 provider_id: "notion",
498 harn_schema_name: "NotionEventPayload",
499 metadata: {
500 let mut metadata = provider_metadata_entry(
501 "notion",
502 &["webhook", "poll"],
503 "NotionEventPayload",
504 &[],
505 hmac_signature_metadata(
506 "notion",
507 "X-Notion-Signature",
508 None,
509 None,
510 None,
511 "hex",
512 ),
513 vec![required_secret("verification_token", "notion")],
514 ProviderRuntimeMetadata::Placeholder,
515 );
516 metadata.outbound_methods = vec![
517 outbound_method("get_page"),
518 outbound_method("update_page"),
519 outbound_method("append_blocks"),
520 outbound_method("query_database"),
521 outbound_method("search"),
522 outbound_method("create_comment"),
523 outbound_method("api_call"),
524 ];
525 metadata
526 },
527 normalize: notion_payload,
528 }),
529 Arc::new(BuiltinProviderSchema {
530 provider_id: "cron",
531 harn_schema_name: "CronEventPayload",
532 metadata: provider_metadata_entry(
533 "cron",
534 &["cron"],
535 "CronEventPayload",
536 &[],
537 SignatureVerificationMetadata::None,
538 Vec::new(),
539 ProviderRuntimeMetadata::Builtin {
540 connector: "cron".to_string(),
541 default_signature_variant: None,
542 },
543 ),
544 normalize: cron_payload,
545 }),
546 Arc::new(BuiltinProviderSchema {
547 provider_id: "webhook",
548 harn_schema_name: "GenericWebhookPayload",
549 metadata: provider_metadata_entry(
550 "webhook",
551 &["webhook"],
552 "GenericWebhookPayload",
553 &[],
554 hmac_signature_metadata(
555 "standard",
556 "webhook-signature",
557 Some("webhook-timestamp"),
558 Some("webhook-id"),
559 Some(300),
560 "base64",
561 ),
562 vec![required_secret("signing_secret", "webhook")],
563 ProviderRuntimeMetadata::Builtin {
564 connector: "webhook".to_string(),
565 default_signature_variant: Some("standard".to_string()),
566 },
567 ),
568 normalize: webhook_payload,
569 }),
570 Arc::new(BuiltinProviderSchema {
571 provider_id: "a2a-push",
572 harn_schema_name: "A2aPushPayload",
573 metadata: provider_metadata_entry(
574 "a2a-push",
575 &["a2a-push"],
576 "A2aPushPayload",
577 &[],
578 SignatureVerificationMetadata::None,
579 Vec::new(),
580 ProviderRuntimeMetadata::Builtin {
581 connector: "a2a-push".to_string(),
582 default_signature_variant: None,
583 },
584 ),
585 normalize: a2a_push_payload,
586 }),
587 Arc::new(stream_provider_schema("kafka", kafka_payload)),
588 Arc::new(stream_provider_schema("nats", nats_payload)),
589 Arc::new(stream_provider_schema("pulsar", pulsar_payload)),
590 Arc::new(stream_provider_schema("postgres-cdc", postgres_cdc_payload)),
591 Arc::new(stream_provider_schema("email", email_payload)),
592 Arc::new(stream_provider_schema("websocket", websocket_payload)),
593 ]
594}
595
596fn stream_provider_schema(
597 provider_id: &'static str,
598 normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
599) -> BuiltinProviderSchema {
600 BuiltinProviderSchema {
601 provider_id,
602 harn_schema_name: "StreamEventPayload",
603 metadata: provider_metadata_entry(
604 provider_id,
605 &["stream"],
606 "StreamEventPayload",
607 &[],
608 SignatureVerificationMetadata::None,
609 Vec::new(),
610 ProviderRuntimeMetadata::Builtin {
611 connector: "stream".to_string(),
612 default_signature_variant: None,
613 },
614 ),
615 normalize,
616 }
617}