fakecloud_eventbridge/
delivery.rs

1use std::sync::Arc;
2
3use chrono::Utc;
4
5use fakecloud_core::delivery::{DeliveryBus, EventBridgeDelivery};
6use fakecloud_lambda::runtime::ContainerRuntime;
7use fakecloud_lambda::SharedLambdaState;
8use fakecloud_logs::SharedLogsState;
9
10use crate::service::{dispatch_event_target, matches_pattern, EventDispatchContext};
11use crate::state::{PutEvent, SharedEventBridgeState};
12
13/// Implements EventBridgeDelivery so other services (SES) can put events
14/// on an EventBridge bus with full rule matching and target delivery.
15pub struct EventBridgeDeliveryImpl {
16    state: SharedEventBridgeState,
17    delivery: Arc<DeliveryBus>,
18    lambda_state: Option<SharedLambdaState>,
19    logs_state: Option<SharedLogsState>,
20    logs_persist: Option<fakecloud_persistence::SnapshotHook>,
21    container_runtime: Option<Arc<ContainerRuntime>>,
22}
23
24impl EventBridgeDeliveryImpl {
25    pub fn new(state: SharedEventBridgeState, delivery: Arc<DeliveryBus>) -> Self {
26        Self {
27            state,
28            delivery,
29            lambda_state: None,
30            logs_state: None,
31            logs_persist: None,
32            container_runtime: None,
33        }
34    }
35
36    pub fn with_lambda(mut self, lambda_state: SharedLambdaState) -> Self {
37        self.lambda_state = Some(lambda_state);
38        self
39    }
40
41    pub fn with_logs(mut self, logs_state: SharedLogsState) -> Self {
42        self.logs_state = Some(logs_state);
43        self
44    }
45
46    /// Wire the CloudWatch Logs persist hook so events this bus delivers to a
47    /// Logs target are written through to the Logs snapshot (see
48    /// `EventDispatchContext::logs_persist`).
49    pub fn with_logs_persist(mut self, hook: fakecloud_persistence::SnapshotHook) -> Self {
50        self.logs_persist = Some(hook);
51        self
52    }
53
54    pub fn with_runtime(mut self, runtime: Arc<ContainerRuntime>) -> Self {
55        self.container_runtime = Some(runtime);
56        self
57    }
58}
59
60impl EventBridgeDeliveryImpl {
61    fn put_event_in_account(
62        &self,
63        source: &str,
64        detail_type: &str,
65        detail: &str,
66        event_bus_name: &str,
67        target_account_id: Option<&str>,
68    ) {
69        let event_id = uuid::Uuid::new_v4().to_string();
70        let now = Utc::now();
71
72        let event = PutEvent {
73            event_id: event_id.clone(),
74            source: source.to_string(),
75            detail_type: detail_type.to_string(),
76            detail: detail.to_string(),
77            event_bus_name: event_bus_name.to_string(),
78            time: now,
79            resources: Vec::new(),
80        };
81
82        let mut accounts = self.state.write();
83        let state = match target_account_id {
84            Some(account_id) if !account_id.is_empty() => accounts.get_or_create(account_id),
85            _ => accounts.default_mut(),
86        };
87        state.events.push(event);
88
89        // Find matching rules and their targets
90        let account_id = state.account_id.clone();
91        let region = state.region.clone();
92        let matching_targets: Vec<(String, crate::state::EventTarget)> = state
93            .rules
94            .values()
95            .filter(|r| {
96                r.event_bus_name == event_bus_name
97                    && r.state == "ENABLED"
98                    && matches_pattern(
99                        r.event_pattern.as_deref(),
100                        source,
101                        detail_type,
102                        detail,
103                        &account_id,
104                        &region,
105                        &[],
106                        &event_id,
107                        &now.to_rfc3339(),
108                    )
109            })
110            .flat_map(|r| r.targets.iter().map(|t| (r.arn.clone(), t.clone())))
111            .collect();
112
113        // Drop the lock before delivering
114        drop(accounts);
115
116        if matching_targets.is_empty() {
117            return;
118        }
119
120        // Build the EventBridge event envelope
121        let detail_value: serde_json::Value =
122            serde_json::from_str(detail).unwrap_or(serde_json::json!({}));
123        let event_json = serde_json::json!({
124            "version": "0",
125            "id": event_id,
126            "source": source,
127            "account": account_id,
128            "detail-type": detail_type,
129            "detail": detail_value,
130            "time": now.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
131            "region": region,
132            "resources": [],
133        });
134        let event_str = event_json.to_string();
135
136        let _ = event_str;
137        let resolved_account = if let Some(acct) = target_account_id {
138            acct.to_string()
139        } else {
140            account_id.clone()
141        };
142        let ctx = EventDispatchContext {
143            state: &self.state,
144            delivery: &self.delivery,
145            lambda_state: self.lambda_state.as_ref(),
146            logs_state: self.logs_state.as_ref(),
147            logs_persist: self.logs_persist.as_ref(),
148            container_runtime: &self.container_runtime,
149            account_id: &resolved_account,
150            region: &region,
151        };
152        for (rule_arn, target) in matching_targets {
153            dispatch_event_target(
154                &ctx,
155                &target,
156                &event_json,
157                &event_id,
158                detail_type,
159                Some(&rule_arn),
160            );
161        }
162    }
163}
164
165impl EventBridgeDelivery for EventBridgeDeliveryImpl {
166    fn put_event(&self, source: &str, detail_type: &str, detail: &str, event_bus_name: &str) {
167        self.put_event_in_account(source, detail_type, detail, event_bus_name, None);
168    }
169
170    fn put_event_to_account(
171        &self,
172        source: &str,
173        detail_type: &str,
174        detail: &str,
175        event_bus_name: &str,
176        target_account_id: &str,
177    ) {
178        self.put_event_in_account(
179            source,
180            detail_type,
181            detail,
182            event_bus_name,
183            Some(target_account_id),
184        );
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191    use crate::state::{EventRule, EventTarget as EbTarget, SharedEventBridgeState};
192    use fakecloud_aws::arn::Arn;
193    use fakecloud_core::delivery::{SnsDelivery, SqsDelivery};
194    use parking_lot::RwLock;
195    use std::collections::{BTreeMap, HashMap};
196    use std::sync::Mutex;
197
198    #[derive(Default)]
199    struct Recorder {
200        sqs: Mutex<Vec<(String, String)>>,
201        sns: Mutex<Vec<(String, String, Option<String>)>>,
202    }
203
204    impl SqsDelivery for Recorder {
205        fn deliver_to_queue(&self, arn: &str, body: &str, _: &HashMap<String, String>) {
206            self.sqs
207                .lock()
208                .unwrap()
209                .push((arn.to_string(), body.to_string()));
210        }
211        fn deliver_to_queue_with_attrs(
212            &self,
213            arn: &str,
214            body: &str,
215            _: &HashMap<String, fakecloud_core::delivery::SqsMessageAttribute>,
216            _: Option<&str>,
217            _: Option<&str>,
218        ) {
219            self.sqs
220                .lock()
221                .unwrap()
222                .push((arn.to_string(), body.to_string()));
223        }
224    }
225
226    impl SnsDelivery for Recorder {
227        fn publish_to_topic(&self, arn: &str, msg: &str, subject: Option<&str>) {
228            self.sns.lock().unwrap().push((
229                arn.to_string(),
230                msg.to_string(),
231                subject.map(|s| s.to_string()),
232            ));
233        }
234    }
235
236    fn make_shared() -> SharedEventBridgeState {
237        Arc::new(RwLock::new(
238            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
239        ))
240    }
241
242    fn make_rule(name: &str, pattern: Option<&str>, target_arn: &str) -> EventRule {
243        EventRule {
244            name: name.to_string(),
245            arn: Arn::new(
246                "events",
247                "us-east-1",
248                "123456789012",
249                &format!("rule/{name}"),
250            )
251            .to_string(),
252            event_bus_name: "default".to_string(),
253            event_pattern: pattern.map(|s| s.to_string()),
254            schedule_expression: None,
255            state: "ENABLED".to_string(),
256            description: None,
257            role_arn: None,
258            managed_by: None,
259            created_by: None,
260            targets: vec![EbTarget {
261                id: "t1".to_string(),
262                arn: target_arn.to_string(),
263                input: None,
264                input_path: None,
265                input_transformer: None,
266                sqs_parameters: None,
267                ..Default::default()
268            }],
269            tags: BTreeMap::new(),
270            last_fired: None,
271        }
272    }
273
274    #[test]
275    fn put_event_appends_to_events_log() {
276        let state = make_shared();
277        let bus = Arc::new(DeliveryBus::new());
278        let delivery = EventBridgeDeliveryImpl::new(state.clone(), bus);
279        delivery.put_event("my.source", "MyType", r#"{"k":"v"}"#, "default");
280        let guard = state.read();
281        let default = guard.default_ref();
282        assert_eq!(default.events.len(), 1);
283        assert_eq!(default.events[0].source, "my.source");
284        assert_eq!(default.events[0].detail_type, "MyType");
285    }
286
287    #[test]
288    fn put_event_dispatches_matching_sqs_target() {
289        let state = make_shared();
290        let q_arn = "arn:aws:sqs:us-east-1:123456789012:q".to_string();
291        {
292            let mut s_accounts = state.write();
293            let s = s_accounts.default_mut();
294            let rule = make_rule("r", None, &q_arn);
295            s.rules
296                .insert(("default".to_string(), "r".to_string()), rule);
297        }
298        let recorder = Arc::new(Recorder::default());
299        let bus = Arc::new(DeliveryBus::new().with_sqs(recorder.clone()));
300        let delivery = EventBridgeDeliveryImpl::new(state, bus);
301        delivery.put_event("app", "Changed", r#"{"x":1}"#, "default");
302        let calls = recorder.sqs.lock().unwrap();
303        assert_eq!(calls.len(), 1);
304        assert_eq!(calls[0].0, q_arn);
305        let env: serde_json::Value = serde_json::from_str(&calls[0].1).unwrap();
306        assert_eq!(env["detail-type"], "Changed");
307        assert_eq!(env["source"], "app");
308    }
309
310    #[test]
311    fn put_event_dispatches_to_sns_target() {
312        let state = make_shared();
313        let topic_arn = "arn:aws:sns:us-east-1:123456789012:t".to_string();
314        {
315            let mut s_accounts = state.write();
316            let s = s_accounts.default_mut();
317            let rule = make_rule("r", None, &topic_arn);
318            s.rules
319                .insert(("default".to_string(), "r".to_string()), rule);
320        }
321        let recorder = Arc::new(Recorder::default());
322        let bus = Arc::new(DeliveryBus::new().with_sns(recorder.clone()));
323        let delivery = EventBridgeDeliveryImpl::new(state, bus);
324        delivery.put_event("app", "Changed", r#"{}"#, "default");
325        let calls = recorder.sns.lock().unwrap();
326        assert_eq!(calls.len(), 1);
327        assert_eq!(calls[0].0, topic_arn);
328        assert_eq!(calls[0].2.as_deref(), Some("Changed"));
329    }
330
331    #[test]
332    fn put_event_skips_disabled_rule() {
333        let state = make_shared();
334        let q_arn = "arn:aws:sqs:us-east-1:123456789012:q".to_string();
335        {
336            let mut s_accounts = state.write();
337            let s = s_accounts.default_mut();
338            let mut rule = make_rule("r", None, &q_arn);
339            rule.state = "DISABLED".to_string();
340            s.rules
341                .insert(("default".to_string(), "r".to_string()), rule);
342        }
343        let recorder = Arc::new(Recorder::default());
344        let bus = Arc::new(DeliveryBus::new().with_sqs(recorder.clone()));
345        let delivery = EventBridgeDeliveryImpl::new(state, bus);
346        delivery.put_event("app", "Changed", r#"{}"#, "default");
347        assert!(recorder.sqs.lock().unwrap().is_empty());
348    }
349
350    #[test]
351    fn put_event_skips_other_bus_rule() {
352        let state = make_shared();
353        let q_arn = "arn:aws:sqs:us-east-1:123456789012:q".to_string();
354        {
355            let mut s_accounts = state.write();
356            let s = s_accounts.default_mut();
357            let mut rule = make_rule("r", None, &q_arn);
358            rule.event_bus_name = "custom-bus".to_string();
359            s.rules
360                .insert(("custom-bus".to_string(), "r".to_string()), rule);
361        }
362        let recorder = Arc::new(Recorder::default());
363        let bus = Arc::new(DeliveryBus::new().with_sqs(recorder.clone()));
364        let delivery = EventBridgeDeliveryImpl::new(state, bus);
365        delivery.put_event("app", "Changed", r#"{}"#, "default");
366        assert!(recorder.sqs.lock().unwrap().is_empty());
367    }
368
369    #[test]
370    fn put_event_handles_invalid_detail_json_gracefully() {
371        let state = make_shared();
372        let q_arn = "arn:aws:sqs:us-east-1:123456789012:q".to_string();
373        {
374            let mut s_accounts = state.write();
375            let s = s_accounts.default_mut();
376            let rule = make_rule("r", None, &q_arn);
377            s.rules
378                .insert(("default".to_string(), "r".to_string()), rule);
379        }
380        let recorder = Arc::new(Recorder::default());
381        let bus = Arc::new(DeliveryBus::new().with_sqs(recorder.clone()));
382        let delivery = EventBridgeDeliveryImpl::new(state, bus);
383        delivery.put_event("app", "Type", "not-json", "default");
384        let calls = recorder.sqs.lock().unwrap();
385        assert_eq!(calls.len(), 1);
386        let env: serde_json::Value = serde_json::from_str(&calls[0].1).unwrap();
387        assert_eq!(env["detail"], serde_json::json!({}));
388    }
389
390    #[test]
391    fn put_event_to_account_writes_to_target_account_bus() {
392        let state = make_shared();
393        let bus = Arc::new(DeliveryBus::new());
394        let delivery = EventBridgeDeliveryImpl::new(state.clone(), bus);
395        delivery.put_event_to_account("scheduler", "Fired", r#"{}"#, "default", "999988887777");
396
397        let guard = state.read();
398        let target = guard
399            .get("999988887777")
400            .expect("target account should be created on demand");
401        assert_eq!(target.events.len(), 1);
402        assert_eq!(target.events[0].source, "scheduler");
403        // The default account's bus should be untouched.
404        assert!(guard.default_ref().events.is_empty());
405    }
406
407    #[test]
408    fn put_event_to_account_dispatches_to_rules_in_target_account() {
409        let state = make_shared();
410        let q_arn = "arn:aws:sqs:us-east-1:999988887777:cross-q".to_string();
411        {
412            let mut s_accounts = state.write();
413            let s = s_accounts.get_or_create("999988887777");
414            let rule = make_rule("xacct-rule", None, &q_arn);
415            s.rules
416                .insert(("default".to_string(), "xacct-rule".to_string()), rule);
417        }
418        let recorder = Arc::new(Recorder::default());
419        let bus = Arc::new(DeliveryBus::new().with_sqs(recorder.clone()));
420        let delivery = EventBridgeDeliveryImpl::new(state, bus);
421        delivery.put_event_to_account(
422            "scheduler",
423            "Cross",
424            r#"{"hi":1}"#,
425            "default",
426            "999988887777",
427        );
428        let calls = recorder.sqs.lock().unwrap();
429        assert_eq!(calls.len(), 1);
430        assert_eq!(calls[0].0, q_arn);
431    }
432}
fakecloud_eventbridge/delivery.rs

fakecloud_eventbridge/
delivery.rs
