harn_vm/triggers/event/
catalog.rs

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, kafka_payload, nats_payload,
10    postgres_cdc_payload, pulsar_payload, webhook_payload, websocket_payload,
11};
12use super::payloads::ProviderPayload;
13
14impl ProviderPayload {
15    pub fn normalize(
16        provider: &ProviderId,
17        kind: &str,
18        headers: &BTreeMap<String, String>,
19        raw: JsonValue,
20    ) -> Result<Self, ProviderCatalogError> {
21        provider_catalog()
22            .read()
23            .expect("provider catalog poisoned")
24            .normalize(provider, kind, headers, raw)
25    }
26}
27
28#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
29pub struct ProviderSecretRequirement {
30    pub name: String,
31    pub required: bool,
32    pub namespace: String,
33}
34
35#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
36pub struct ProviderOutboundMethod {
37    pub name: String,
38}
39
40#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
41#[serde(tag = "kind", rename_all = "snake_case")]
42pub enum SignatureVerificationMetadata {
43    #[default]
44    None,
45    Hmac {
46        variant: String,
47        raw_body: bool,
48        signature_header: String,
49        timestamp_header: Option<String>,
50        id_header: Option<String>,
51        default_tolerance_secs: Option<i64>,
52        digest: String,
53        encoding: String,
54    },
55}
56
57#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
58#[serde(tag = "kind", rename_all = "snake_case")]
59pub enum ProviderRuntimeMetadata {
60    Builtin {
61        connector: String,
62        default_signature_variant: Option<String>,
63    },
64    #[default]
65    Placeholder,
66}
67
68#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
69pub struct ProviderMetadata {
70    pub provider: String,
71    #[serde(default)]
72    pub kinds: Vec<String>,
73    pub schema_name: String,
74    #[serde(default)]
75    pub outbound_methods: Vec<ProviderOutboundMethod>,
76    #[serde(default)]
77    pub secret_requirements: Vec<ProviderSecretRequirement>,
78    #[serde(default)]
79    pub signature_verification: SignatureVerificationMetadata,
80    #[serde(default)]
81    pub runtime: ProviderRuntimeMetadata,
82}
83
84impl ProviderMetadata {
85    pub fn supports_kind(&self, kind: &str) -> bool {
86        self.kinds.iter().any(|candidate| candidate == kind)
87    }
88
89    pub fn required_secret_names(&self) -> impl Iterator<Item = &str> {
90        self.secret_requirements
91            .iter()
92            .filter(|requirement| requirement.required)
93            .map(|requirement| requirement.name.as_str())
94    }
95}
96
97pub trait ProviderSchema: Send + Sync {
98    fn provider_id(&self) -> &str;
99    fn harn_schema_name(&self) -> &str;
100    fn metadata(&self) -> ProviderMetadata {
101        ProviderMetadata {
102            provider: self.provider_id().to_string(),
103            schema_name: self.harn_schema_name().to_string(),
104            ..ProviderMetadata::default()
105        }
106    }
107    fn normalize(
108        &self,
109        kind: &str,
110        headers: &BTreeMap<String, String>,
111        raw: JsonValue,
112    ) -> Result<ProviderPayload, ProviderCatalogError>;
113}
114
115#[derive(Clone, Debug, PartialEq, Eq)]
116pub enum ProviderCatalogError {
117    DuplicateProvider(String),
118    UnknownProvider(String),
119}
120
121impl std::fmt::Display for ProviderCatalogError {
122    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123        match self {
124            Self::DuplicateProvider(provider) => {
125                write!(f, "provider `{provider}` is already registered")
126            }
127            Self::UnknownProvider(provider) => write!(f, "provider `{provider}` is not registered"),
128        }
129    }
130}
131
132impl std::error::Error for ProviderCatalogError {}
133
134#[derive(Clone, Default)]
135pub struct ProviderCatalog {
136    providers: BTreeMap<String, Arc<dyn ProviderSchema>>,
137}
138
139impl ProviderCatalog {
140    pub fn with_defaults() -> Self {
141        let mut catalog = Self::default();
142        for schema in default_provider_schemas() {
143            catalog
144                .register(schema)
145                .expect("default providers must register cleanly");
146        }
147        catalog
148    }
149
150    pub fn with_defaults_and(
151        schemas: Vec<Arc<dyn ProviderSchema>>,
152    ) -> Result<Self, ProviderCatalogError> {
153        let mut catalog = Self::with_defaults();
154        catalog.merge(schemas)?;
155        Ok(catalog)
156    }
157
158    /// Add `schemas` to this catalog without disturbing what is already in it.
159    ///
160    /// A contribution is skipped when its id belongs to a core schema — a
161    /// package cannot redefine `webhook` — and when it re-registers a provider
162    /// under the schema name already recorded for it, which is what reloading
163    /// the same manifest looks like. Anything else is a real disagreement
164    /// about what a provider id means, and is reported rather than settled by
165    /// load order.
166    pub fn merge(
167        &mut self,
168        schemas: Vec<Arc<dyn ProviderSchema>>,
169    ) -> Result<(), ProviderCatalogError> {
170        for schema in schemas {
171            let provider = schema.provider_id().to_string();
172            match self.providers.get(provider.as_str()) {
173                Some(existing)
174                    if default_provider_ids().contains(provider.as_str())
175                        || existing.harn_schema_name() == schema.harn_schema_name() => {}
176                Some(_) => return Err(ProviderCatalogError::DuplicateProvider(provider)),
177                None => {
178                    self.providers.insert(provider, schema);
179                }
180            }
181        }
182        Ok(())
183    }
184
185    pub fn register(
186        &mut self,
187        schema: Arc<dyn ProviderSchema>,
188    ) -> Result<(), ProviderCatalogError> {
189        let provider = schema.provider_id().to_string();
190        if self.providers.contains_key(provider.as_str()) {
191            return Err(ProviderCatalogError::DuplicateProvider(provider));
192        }
193        self.providers.insert(provider, schema);
194        Ok(())
195    }
196
197    pub fn normalize(
198        &self,
199        provider: &ProviderId,
200        kind: &str,
201        headers: &BTreeMap<String, String>,
202        raw: JsonValue,
203    ) -> Result<ProviderPayload, ProviderCatalogError> {
204        let schema = self
205            .providers
206            .get(provider.as_str())
207            .ok_or_else(|| ProviderCatalogError::UnknownProvider(provider.0.clone()))?;
208        schema.normalize(kind, headers, raw)
209    }
210
211    pub fn schema_names(&self) -> BTreeMap<String, String> {
212        self.providers
213            .iter()
214            .map(|(provider, schema)| (provider.clone(), schema.harn_schema_name().to_string()))
215            .collect()
216    }
217
218    pub fn entries(&self) -> Vec<ProviderMetadata> {
219        self.providers
220            .values()
221            .map(|schema| schema.metadata())
222            .collect()
223    }
224
225    pub fn metadata_for(&self, provider: &str) -> Option<ProviderMetadata> {
226        self.providers.get(provider).map(|schema| schema.metadata())
227    }
228}
229
230/// Contribute `schemas` to the process-wide catalog.
231///
232/// Loading a package's runtime extensions says what that package provides; it
233/// does not describe the whole world. Components that load packages
234/// independently — an orchestrator harness and a persona command sharing a
235/// process — therefore compose rather than erase each other's providers.
236pub fn register_provider_schemas(
237    schemas: Vec<Arc<dyn ProviderSchema>>,
238) -> Result<(), ProviderCatalogError> {
239    provider_catalog()
240        .write()
241        .expect("provider catalog poisoned")
242        .merge(schemas)
243}
244
245/// Drop every contributed provider, leaving the builtin schemas.
246pub fn reset_provider_catalog() {
247    *provider_catalog()
248        .write()
249        .expect("provider catalog poisoned") = ProviderCatalog::with_defaults();
250}
251
252pub fn registered_provider_schema_names() -> BTreeMap<String, String> {
253    provider_catalog()
254        .read()
255        .expect("provider catalog poisoned")
256        .schema_names()
257}
258
259pub fn registered_provider_metadata() -> Vec<ProviderMetadata> {
260    provider_catalog()
261        .read()
262        .expect("provider catalog poisoned")
263        .entries()
264}
265
266pub fn provider_metadata(provider: &str) -> Option<ProviderMetadata> {
267    provider_catalog()
268        .read()
269        .expect("provider catalog poisoned")
270        .metadata_for(provider)
271}
272
273fn provider_catalog() -> &'static RwLock<ProviderCatalog> {
274    static PROVIDER_CATALOG: OnceLock<RwLock<ProviderCatalog>> = OnceLock::new();
275    PROVIDER_CATALOG.get_or_init(|| RwLock::new(ProviderCatalog::with_defaults()))
276}
277
278fn default_provider_ids() -> &'static BTreeSet<String> {
279    static DEFAULT_PROVIDER_IDS: OnceLock<BTreeSet<String>> = OnceLock::new();
280    DEFAULT_PROVIDER_IDS.get_or_init(|| {
281        default_provider_schemas()
282            .iter()
283            .map(|schema| schema.provider_id().to_string())
284            .collect()
285    })
286}
287
288struct BuiltinProviderSchema {
289    provider_id: &'static str,
290    harn_schema_name: &'static str,
291    metadata: ProviderMetadata,
292    normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
293}
294
295impl ProviderSchema for BuiltinProviderSchema {
296    fn provider_id(&self) -> &str {
297        self.provider_id
298    }
299
300    fn harn_schema_name(&self) -> &str {
301        self.harn_schema_name
302    }
303
304    fn metadata(&self) -> ProviderMetadata {
305        self.metadata.clone()
306    }
307
308    fn normalize(
309        &self,
310        kind: &str,
311        headers: &BTreeMap<String, String>,
312        raw: JsonValue,
313    ) -> Result<ProviderPayload, ProviderCatalogError> {
314        Ok((self.normalize)(kind, headers, raw))
315    }
316}
317
318fn provider_metadata_entry(
319    provider: &str,
320    kinds: &[&str],
321    schema_name: &str,
322    outbound_methods: &[&str],
323    signature_verification: SignatureVerificationMetadata,
324    secret_requirements: Vec<ProviderSecretRequirement>,
325    runtime: ProviderRuntimeMetadata,
326) -> ProviderMetadata {
327    ProviderMetadata {
328        provider: provider.to_string(),
329        kinds: kinds.iter().map(|kind| kind.to_string()).collect(),
330        schema_name: schema_name.to_string(),
331        outbound_methods: outbound_methods
332            .iter()
333            .map(|name| ProviderOutboundMethod {
334                name: (*name).to_string(),
335            })
336            .collect(),
337        secret_requirements,
338        signature_verification,
339        runtime,
340    }
341}
342
343fn hmac_signature_metadata(
344    variant: &str,
345    signature_header: &str,
346    timestamp_header: Option<&str>,
347    id_header: Option<&str>,
348    default_tolerance_secs: Option<i64>,
349    encoding: &str,
350) -> SignatureVerificationMetadata {
351    SignatureVerificationMetadata::Hmac {
352        variant: variant.to_string(),
353        raw_body: true,
354        signature_header: signature_header.to_string(),
355        timestamp_header: timestamp_header.map(ToString::to_string),
356        id_header: id_header.map(ToString::to_string),
357        default_tolerance_secs,
358        digest: "sha256".to_string(),
359        encoding: encoding.to_string(),
360    }
361}
362
363fn required_secret(name: &str, namespace: &str) -> ProviderSecretRequirement {
364    ProviderSecretRequirement {
365        name: name.to_string(),
366        required: true,
367        namespace: namespace.to_string(),
368    }
369}
370
371fn default_provider_schemas() -> Vec<Arc<dyn ProviderSchema>> {
372    vec![
373        Arc::new(BuiltinProviderSchema {
374            provider_id: "cron",
375            harn_schema_name: "CronEventPayload",
376            metadata: provider_metadata_entry(
377                "cron",
378                &["cron"],
379                "CronEventPayload",
380                &[],
381                SignatureVerificationMetadata::None,
382                Vec::new(),
383                ProviderRuntimeMetadata::Builtin {
384                    connector: "cron".to_string(),
385                    default_signature_variant: None,
386                },
387            ),
388            normalize: cron_payload,
389        }),
390        Arc::new(BuiltinProviderSchema {
391            provider_id: "webhook",
392            harn_schema_name: "GenericWebhookPayload",
393            metadata: provider_metadata_entry(
394                "webhook",
395                &["webhook"],
396                "GenericWebhookPayload",
397                &[],
398                hmac_signature_metadata(
399                    "standard",
400                    "webhook-signature",
401                    Some("webhook-timestamp"),
402                    Some("webhook-id"),
403                    Some(300),
404                    "base64",
405                ),
406                vec![required_secret("signing_secret", "webhook")],
407                ProviderRuntimeMetadata::Builtin {
408                    connector: "webhook".to_string(),
409                    default_signature_variant: Some("standard".to_string()),
410                },
411            ),
412            normalize: webhook_payload,
413        }),
414        Arc::new(BuiltinProviderSchema {
415            provider_id: "a2a-push",
416            harn_schema_name: "A2aPushPayload",
417            metadata: provider_metadata_entry(
418                "a2a-push",
419                &["a2a-push"],
420                "A2aPushPayload",
421                &[],
422                SignatureVerificationMetadata::None,
423                Vec::new(),
424                ProviderRuntimeMetadata::Builtin {
425                    connector: "a2a-push".to_string(),
426                    default_signature_variant: None,
427                },
428            ),
429            normalize: a2a_push_payload,
430        }),
431        Arc::new(stream_provider_schema("kafka", kafka_payload)),
432        Arc::new(stream_provider_schema("nats", nats_payload)),
433        Arc::new(stream_provider_schema("pulsar", pulsar_payload)),
434        Arc::new(stream_provider_schema("postgres-cdc", postgres_cdc_payload)),
435        Arc::new(stream_provider_schema("email", email_payload)),
436        Arc::new(stream_provider_schema("websocket", websocket_payload)),
437    ]
438}
439
440fn stream_provider_schema(
441    provider_id: &'static str,
442    normalize: fn(&str, &BTreeMap<String, String>, JsonValue) -> ProviderPayload,
443) -> BuiltinProviderSchema {
444    BuiltinProviderSchema {
445        provider_id,
446        harn_schema_name: "StreamEventPayload",
447        metadata: provider_metadata_entry(
448            provider_id,
449            &["stream"],
450            "StreamEventPayload",
451            &[],
452            SignatureVerificationMetadata::None,
453            Vec::new(),
454            ProviderRuntimeMetadata::Builtin {
455                connector: "stream".to_string(),
456                default_signature_variant: None,
457            },
458        ),
459        normalize,
460    }
461}
harn_vm/triggers/event/catalog.rs

harn_vm/triggers/event/
catalog.rs
