fakecloud_core/
delivery.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3
4/// Cross-service message delivery.
5///
6/// Services use this to deliver messages to other services without
7/// direct dependencies between service crates. The server wires up
8/// the delivery functions at startup.
9pub struct DeliveryBus {
10    /// Deliver a message to an SQS queue by ARN.
11    sqs_sender: Option<Arc<dyn SqsDelivery>>,
12    /// Publish a message to an SNS topic by ARN.
13    sns_sender: Option<Arc<dyn SnsDelivery>>,
14    /// Put an event onto an EventBridge bus.
15    eventbridge_sender: Option<Arc<dyn EventBridgeDelivery>>,
16    /// Invoke a Lambda function by ARN.
17    lambda_invoker: Option<Arc<dyn LambdaDelivery>>,
18    /// Put records to a Kinesis Data Stream by ARN.
19    kinesis_sender: Option<Arc<dyn KinesisDelivery>>,
20    /// Start Step Functions executions.
21    stepfunctions_starter: Option<Arc<dyn StepFunctionsDelivery>>,
22    /// Start SageMaker pipeline executions (Scheduler sagemaker target).
23    sagemaker_pipeline_starter: Option<Arc<dyn SageMakerPipelineDelivery>>,
24    /// Write objects to S3 buckets.
25    s3_writer: Option<Arc<dyn S3Delivery>>,
26    /// Put records into Firehose delivery streams.
27    firehose_sender: Option<Arc<dyn FirehoseDelivery>>,
28    /// Send a high-level SES SendEmail call (cross-service universal target).
29    ses_dispatcher: Option<Arc<dyn SesSendEmailDispatcher>>,
30    /// Run an ECS task on a cluster (cross-service universal target).
31    ecs_task_runner: Option<Arc<dyn EcsTaskRunner>>,
32    /// Register/deregister ELBv2 targets from ECS runtime.
33    elbv2_target_registration: Option<Arc<dyn Elbv2TargetRegistration>>,
34    /// Publish CloudWatch metric data points (CloudWatch Logs metric
35    /// filters extract these on PutLogEvents).
36    cloudwatch_metrics: Option<Arc<dyn CloudwatchDelivery>>,
37    /// Put log events into CloudWatch Logs log groups. Used by Step
38    /// Functions Express execution logging and ECS awslogs driver.
39    cloudwatch_logs: Option<Arc<dyn CloudwatchLogsDelivery>>,
40    /// Verify a Cognito-issued JWT against the user pool that issued it.
41    /// Used by API Gateway v1's `COGNITO_USER_POOLS` authorizer to
42    /// validate signature/expiry/audience and extract claims.
43    cognito_jwt_verifier: Option<Arc<dyn CognitoJwtVerifier>>,
44    /// Encrypt/decrypt via KMS for cross-service SSE (S3 SSE-KMS, SES
45    /// S3Action KmsKeyArn, etc.).
46    kms_hook: Option<Arc<dyn KmsHook>>,
47}
48
49/// Message attribute for SQS delivery from SNS.
50#[derive(Debug, Clone)]
51pub struct SqsMessageAttribute {
52    pub data_type: String,
53    pub string_value: Option<String>,
54    pub binary_value: Option<String>,
55}
56
57/// Error returned by fallible SQS delivery. Used by Scheduler's DLQ
58/// routing, which must distinguish "target queue missing" from
59/// "delivered successfully" to decide whether to send to the
60/// `DeadLetterConfig.Arn`.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub enum SqsDeliveryError {
63    /// The target queue ARN did not resolve to any existing queue.
64    QueueNotFound(String),
65    /// The ARN could not be parsed into a valid SQS queue identifier.
66    InvalidArn(String),
67    /// The message violated a constraint required by the target queue
68    /// (e.g. FIFO send missing MessageDeduplicationId without
69    /// content-based dedup enabled). Surfaces as a non-retriable
70    /// failure so the upstream service can route to its configured DLQ.
71    InvalidParameter(String),
72}
73
74impl std::fmt::Display for SqsDeliveryError {
75    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76        match self {
77            Self::QueueNotFound(arn) => write!(f, "queue not found: {arn}"),
78            Self::InvalidArn(arn) => write!(f, "invalid queue ARN: {arn}"),
79            Self::InvalidParameter(msg) => write!(f, "invalid parameter: {msg}"),
80        }
81    }
82}
83
84impl std::error::Error for SqsDeliveryError {}
85
86/// Trait for delivering messages to SQS queues.
87pub trait SqsDelivery: Send + Sync {
88    fn deliver_to_queue(
89        &self,
90        queue_arn: &str,
91        message_body: &str,
92        attributes: &HashMap<String, String>,
93    );
94
95    /// Deliver with message attributes and FIFO fields
96    fn deliver_to_queue_with_attrs(
97        &self,
98        queue_arn: &str,
99        message_body: &str,
100        message_attributes: &HashMap<String, SqsMessageAttribute>,
101        message_group_id: Option<&str>,
102        message_dedup_id: Option<&str>,
103    ) {
104        // Default implementation: fall back to simple delivery
105        let _ = (message_attributes, message_group_id, message_dedup_id);
106        self.deliver_to_queue(queue_arn, message_body, &HashMap::new());
107    }
108
109    /// The stored ARN of the queue a QueueUrl (`<endpoint>/<account>/<name>`)
110    /// names, looked up by account and queue name. A QueueUrl carries no
111    /// region, so callers holding only a URL must not rebuild the ARN from
112    /// their own region. `None` when no such queue exists.
113    fn queue_arn_for_url(&self, queue_url: &str) -> Option<String> {
114        let _ = queue_url;
115        None
116    }
117
118    /// Fallible variant used by Scheduler's DLQ routing. Default
119    /// implementation assumes the queue exists (preserving the
120    /// fire-and-forget semantics of `deliver_to_queue`); the real SQS
121    /// impl overrides this to actually look up the queue and report
122    /// `QueueNotFound` so the caller can route to a DLQ.
123    fn try_deliver_to_queue_with_attrs(
124        &self,
125        queue_arn: &str,
126        message_body: &str,
127        message_attributes: &HashMap<String, SqsMessageAttribute>,
128        message_group_id: Option<&str>,
129        message_dedup_id: Option<&str>,
130    ) -> Result<(), SqsDeliveryError> {
131        self.deliver_to_queue_with_attrs(
132            queue_arn,
133            message_body,
134            message_attributes,
135            message_group_id,
136            message_dedup_id,
137        );
138        Ok(())
139    }
140}
141
142/// Trait for publishing messages to SNS topics.
143pub trait SnsDelivery: Send + Sync {
144    fn publish_to_topic(&self, topic_arn: &str, message: &str, subject: Option<&str>);
145
146    /// Publish to a FIFO SNS topic carrying the message group/dedup IDs
147    /// that downstream SQS subscribers need for ordering. Default impl
148    /// drops the IDs so non-FIFO callers don't have to override.
149    fn publish_to_topic_fifo(
150        &self,
151        topic_arn: &str,
152        message: &str,
153        subject: Option<&str>,
154        _message_group_id: Option<&str>,
155        _message_dedup_id: Option<&str>,
156    ) {
157        self.publish_to_topic(topic_arn, message, subject);
158    }
159}
160
161/// Trait for putting events onto an EventBridge bus from cross-service integrations.
162pub trait EventBridgeDelivery: Send + Sync {
163    /// Put an event onto the specified event bus in the default account.
164    /// The implementation should handle rule matching and target delivery.
165    fn put_event(&self, source: &str, detail_type: &str, detail: &str, event_bus_name: &str);
166
167    /// Put an event onto the specified event bus owned by `target_account_id`.
168    /// Used for cross-account delivery where the source service (e.g. Scheduler)
169    /// has a target ARN containing the destination account. The default impl
170    /// falls back to the default-account `put_event` for backwards compat —
171    /// real implementations should override and route to the target account's
172    /// state.
173    fn put_event_to_account(
174        &self,
175        source: &str,
176        detail_type: &str,
177        detail: &str,
178        event_bus_name: &str,
179        _target_account_id: &str,
180    ) {
181        self.put_event(source, detail_type, detail, event_bus_name);
182    }
183}
184
185/// Trait for invoking Lambda functions from cross-service integrations.
186pub trait LambdaDelivery: Send + Sync {
187    /// Invoke a Lambda function with the given payload.
188    /// The function is identified by ARN. Returns the response bytes on success.
189    fn invoke_lambda(
190        &self,
191        function_arn: &str,
192        payload: &str,
193    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send>>;
194}
195
196/// Trait for putting records to Kinesis Data Streams.
197pub trait KinesisDelivery: Send + Sync {
198    /// Put a record to a Kinesis stream identified by ARN.
199    /// The data should be base64-encoded. partition_key is used for shard distribution.
200    fn put_record(&self, stream_arn: &str, data: &str, partition_key: &str);
201}
202
203/// Trait for starting Step Functions executions from cross-service integrations.
204pub trait StepFunctionsDelivery: Send + Sync {
205    /// Start a state machine execution with the given input.
206    /// The state machine is identified by ARN.
207    fn start_execution(&self, state_machine_arn: &str, input: &str);
208}
209
210/// Cross-service SageMaker pipeline dispatch used by EventBridge Scheduler's
211/// `sagemaker:pipeline` target. The pipeline is identified by ARN
212/// (`arn:aws:sagemaker:<region>:<account>:pipeline/<name>`); `parameters` is the
213/// target's `PipelineParameterList`.
214pub trait SageMakerPipelineDelivery: Send + Sync {
215    fn start_pipeline_execution(&self, pipeline_arn: &str, parameters: &serde_json::Value);
216}
217
218/// Cross-service Kinesis Data Firehose dispatch used by services
219/// (CloudWatch Logs subscription filters, EventBridge targets) that
220/// route records into a delivery stream without depending on the
221/// firehose crate directly. ARN form is
222/// `arn:aws:firehose:<region>:<account>:deliverystream/<name>`.
223pub trait FirehoseDelivery: Send + Sync {
224    fn put_record(&self, delivery_stream_arn: &str, data: &[u8]);
225}
226
227/// Cross-service S3 writer used by services that need to deliver
228/// content to S3 buckets without taking a direct dep on the S3 crate
229/// (CloudWatch Logs export tasks, Kinesis Firehose, ELB access logs).
230pub trait S3Delivery: Send + Sync {
231    /// Put an object to a bucket. Returns the bucket name on success
232    /// or an error string the caller can surface in tests / logs.
233    fn put_object(
234        &self,
235        account_id: &str,
236        bucket: &str,
237        key: &str,
238        body: Vec<u8>,
239        content_type: Option<&str>,
240    ) -> Result<(), String>;
241
242    /// Read an object's body. Returns Err when the bucket or key does
243    /// not exist or when the body cannot be read. Used by RDS
244    /// `RestoreDBInstanceFromS3` to ingest a backup blob without taking
245    /// a direct dep on the S3 crate.
246    fn get_object(&self, account_id: &str, bucket: &str, key: &str) -> Result<Vec<u8>, String>;
247}
248
249/// SES SendEmail dispatch for callers that already speak the AWS SES
250/// SendEmail / SendEmailV2 shape (multiple to/cc/bcc, optional subject,
251/// text/html bodies). Distinct from `EmailDispatcher`, which is the
252/// single-recipient cross-service primitive used by Cognito.
253pub trait SesSendEmailDispatcher: Send + Sync {
254    #[allow(clippy::too_many_arguments)]
255    fn send_email(
256        &self,
257        account_id: &str,
258        from: &str,
259        to: Vec<String>,
260        cc: Vec<String>,
261        bcc: Vec<String>,
262        subject: Option<&str>,
263        text_body: Option<&str>,
264        html_body: Option<&str>,
265    ) -> Result<(), String>;
266}
267
268/// Synthesize an ECS RunTask call from outside the ECS crate. Used by
269/// EventBridge Scheduler and EventBridge Rules to start tasks without
270/// depending directly on `fakecloud_ecs`.
271pub trait EcsTaskRunner: Send + Sync {
272    fn run_task(
273        &self,
274        account_id: &str,
275        cluster: &str,
276        task_definition: &str,
277        launch_type: Option<&str>,
278        count: usize,
279    ) -> Result<(), String>;
280}
281
282/// Register/deregister targets with ELBv2 target groups from outside
283/// the elbv2 crate. Used by ECS runtime when tasks with load balancers
284/// reach RUNNING or STOPPED.
285pub trait Elbv2TargetRegistration: Send + Sync {
286    fn register_targets(
287        &self,
288        account_id: &str,
289        target_group_arn: &str,
290        targets: Vec<(String, Option<i64>)>,
291    );
292    fn deregister_targets(
293        &self,
294        account_id: &str,
295        target_group_arn: &str,
296        targets: Vec<(String, Option<i64>)>,
297    );
298}
299
300/// Publish CloudWatch metric data points from outside the cloudwatch
301/// crate. Used by CloudWatch Logs metric filters when an incoming log
302/// event matches their pattern.
303pub trait CloudwatchDelivery: Send + Sync {
304    #[allow(clippy::too_many_arguments)]
305    fn put_metric(
306        &self,
307        account_id: &str,
308        region: &str,
309        namespace: &str,
310        metric_name: &str,
311        value: f64,
312        unit: Option<&str>,
313        dimensions: std::collections::BTreeMap<String, String>,
314        timestamp_ms: i64,
315    );
316}
317
318/// Put log events into CloudWatch Logs log groups from outside the
319/// logs crate. Used by Step Functions Express execution logging and
320/// ECS awslogs driver so they can deliver without depending directly
321/// on fakecloud_logs.
322pub trait CloudwatchLogsDelivery: Send + Sync {
323    fn put_log_events(
324        &self,
325        account_id: &str,
326        log_group_name: &str,
327        log_stream_name: &str,
328        events: &[(i64, String)],
329    );
330}
331
332/// Outbound email dispatch used by services that emulate AWS flows that
333/// route through SES (Cognito verification, etc.) without taking a direct
334/// dependency on the SES crate.
335pub trait EmailDispatcher: Send + Sync {
336    fn send_email(
337        &self,
338        account_id: &str,
339        from: &str,
340        to: &str,
341        subject: &str,
342        body_text: &str,
343        body_html: Option<&str>,
344    );
345}
346
347/// Outbound SMS dispatch used by services that emulate AWS flows that route
348/// through SNS phone-number publish (Cognito SMS MFA, etc.).
349pub trait SmsDispatcher: Send + Sync {
350    fn send_sms(&self, account_id: &str, phone_number: &str, message: &str);
351}
352
353/// Cross-service KMS hook: services that accept a `KmsKeyId` (Secrets
354/// Manager, SSM `SecureString`, S3 SSE-KMS, SQS / SNS / DynamoDB
355/// encrypted resources) call this so that real KMS calls happen, the
356/// invocation is recorded for introspection, and the returned blob is
357/// decryptable by the public KMS API.
358///
359/// Encryption context is the AWS-defined per-service map (e.g.
360/// `{aws:secretsmanager:secretArn: <arn>}` for Secrets Manager) and is
361/// recorded with the call so test code can assert it.
362pub trait KmsHook: Send + Sync {
363    fn encrypt(
364        &self,
365        account_id: &str,
366        region: &str,
367        key_id: &str,
368        plaintext: &[u8],
369        service_principal: &str,
370        encryption_context: std::collections::HashMap<String, String>,
371    ) -> Result<String, String>;
372
373    fn decrypt(
374        &self,
375        account_id: &str,
376        ciphertext_b64: &str,
377        service_principal: &str,
378        encryption_context: std::collections::HashMap<String, String>,
379    ) -> Result<Vec<u8>, String>;
380
381    /// Resolve `key_id` (key id, key ARN, alias name or alias ARN) to the
382    /// key's ARN, provisioning the AWS-managed `aws/<service>` key on first
383    /// use. Services report the resolved ARN on their resource descriptions.
384    fn resolve_key_arn(
385        &self,
386        _account_id: &str,
387        _region: &str,
388        _key_id: &str,
389        _service_principal: &str,
390    ) -> Result<String, String> {
391        Err("key resolution is not supported by this hook".to_string())
392    }
393
394    /// The ARN of the AWS-managed key for `service` (the key behind
395    /// `alias/aws/<service>`) in `account_id` and `region`, minted on first
396    /// use. AWS-managed keys exist once per account AND region, so each
397    /// region gets its own key in its own partition.
398    fn aws_managed_key_arn(
399        &self,
400        _account_id: &str,
401        _region: &str,
402        _service: &str,
403        _service_principal: &str,
404    ) -> Result<String, String> {
405        Err("AWS-managed key resolution is not supported by this hook".to_string())
406    }
407}
408
409/// The ARN of the AWS-managed KMS key for `service` (the key behind
410/// `alias/aws/<service>`) in `account_id` and `region`, which a service reports
411/// for a resource encrypted at rest when the caller named no key. The key is
412/// resolved through the KMS hook, which mints it on first use in that account
413/// and region (and the region's partition), as real KMS does, so the ARN is a
414/// real key `DescribeKey` / `Decrypt` accept. `None` when no hook is wired (a
415/// service running outside the server) or resolution fails; a failure is
416/// logged, never swallowed.
417pub fn aws_managed_kms_key_arn(
418    hook: Option<&dyn KmsHook>,
419    account_id: &str,
420    region: &str,
421    service: &str,
422) -> Option<String> {
423    let hook = hook?;
424    let alias = format!("alias/aws/{service}");
425    let principal = format!("{service}.amazonaws.com");
426    match hook.aws_managed_key_arn(account_id, region, service, &principal) {
427        Ok(arn) => Some(arn),
428        Err(err) => {
429            tracing::warn!(
430                %err,
431                account_id,
432                region,
433                alias = %alias,
434                "could not resolve the AWS-managed KMS key; reporting no key"
435            );
436            None
437        }
438    }
439}
440
441/// The KMS key a resource is encrypted with: the caller-named key (`named`,
442/// ignored when empty) or, when none is named, the AWS-managed key for
443/// `service` in `account_id` and `region` (see [`aws_managed_kms_key_arn`]).
444pub fn kms_key_or_aws_managed(
445    hook: Option<&dyn KmsHook>,
446    named: Option<&str>,
447    account_id: &str,
448    region: &str,
449    service: &str,
450) -> Option<String> {
451    match named.filter(|k| !k.is_empty()) {
452        Some(key) => Some(key.to_string()),
453        None => aws_managed_kms_key_arn(hook, account_id, region, service),
454    }
455}
456
457/// Cognito-issued JWT verification hook. Implementations are wired by
458/// fakecloud-server and back the `COGNITO_USER_POOLS` authorizer in
459/// API Gateway v1. The verifier validates RS256 signature, exp/nbf,
460/// `iss`, and `aud`/`client_id` against the user pool referenced by
461/// the authorizer's `providerArns`. On success returns the decoded
462/// claims as a JSON object; on failure returns an error string the
463/// caller surfaces as `401 Unauthorized`.
464pub trait CognitoJwtVerifier: Send + Sync {
465    fn verify_token(
466        &self,
467        account_id: &str,
468        user_pool_arn: &str,
469        token: &str,
470    ) -> Result<serde_json::Value, String>;
471}
472
473impl DeliveryBus {
474    pub fn new() -> Self {
475        Self {
476            sqs_sender: None,
477            sns_sender: None,
478            eventbridge_sender: None,
479            lambda_invoker: None,
480            kinesis_sender: None,
481            stepfunctions_starter: None,
482            sagemaker_pipeline_starter: None,
483            s3_writer: None,
484            firehose_sender: None,
485            ses_dispatcher: None,
486            ecs_task_runner: None,
487            elbv2_target_registration: None,
488            cloudwatch_metrics: None,
489            cloudwatch_logs: None,
490            cognito_jwt_verifier: None,
491            kms_hook: None,
492        }
493    }
494
495    pub fn with_cognito_jwt_verifier(mut self, verifier: Arc<dyn CognitoJwtVerifier>) -> Self {
496        self.cognito_jwt_verifier = Some(verifier);
497        self
498    }
499
500    pub fn with_kms_hook(mut self, hook: Arc<dyn KmsHook>) -> Self {
501        self.kms_hook = Some(hook);
502        self
503    }
504
505    /// Encrypt plaintext with the supplied KMS key. Returns Err when no
506    /// KMS hook is wired or the encryption fails.
507    pub fn kms_encrypt(
508        &self,
509        account_id: &str,
510        region: &str,
511        key_id: &str,
512        plaintext: &[u8],
513        service_principal: &str,
514        encryption_context: std::collections::HashMap<String, String>,
515    ) -> Result<String, String> {
516        match self.kms_hook {
517            Some(ref h) => h.encrypt(
518                account_id,
519                region,
520                key_id,
521                plaintext,
522                service_principal,
523                encryption_context,
524            ),
525            None => Err("KMS hook not configured".to_string()),
526        }
527    }
528
529    /// Decrypt a KMS ciphertext blob. Returns Err when no KMS hook is
530    /// wired or the decryption fails.
531    pub fn kms_decrypt(
532        &self,
533        account_id: &str,
534        ciphertext_b64: &str,
535        service_principal: &str,
536        encryption_context: std::collections::HashMap<String, String>,
537    ) -> Result<Vec<u8>, String> {
538        match self.kms_hook {
539            Some(ref h) => h.decrypt(
540                account_id,
541                ciphertext_b64,
542                service_principal,
543                encryption_context,
544            ),
545            None => Err("KMS hook not configured".to_string()),
546        }
547    }
548
549    /// Verify a Cognito JWT against a user pool. Returns `Err` when no
550    /// verifier is wired or when the token fails validation.
551    pub fn verify_cognito_jwt(
552        &self,
553        account_id: &str,
554        user_pool_arn: &str,
555        token: &str,
556    ) -> Result<serde_json::Value, String> {
557        match self.cognito_jwt_verifier {
558            Some(ref v) => v.verify_token(account_id, user_pool_arn, token),
559            None => Err("Cognito JWT verifier not configured".to_string()),
560        }
561    }
562
563    pub fn with_cloudwatch_metrics(mut self, sender: Arc<dyn CloudwatchDelivery>) -> Self {
564        self.cloudwatch_metrics = Some(sender);
565        self
566    }
567
568    /// Publish a CloudWatch metric data point. Silently no-ops when no
569    /// CloudWatch sender is wired (in-process tests not exercising the
570    /// metrics path).
571    #[allow(clippy::too_many_arguments)]
572    pub fn put_cloudwatch_metric(
573        &self,
574        account_id: &str,
575        region: &str,
576        namespace: &str,
577        metric_name: &str,
578        value: f64,
579        unit: Option<&str>,
580        dimensions: std::collections::BTreeMap<String, String>,
581        timestamp_ms: i64,
582    ) {
583        if let Some(ref sender) = self.cloudwatch_metrics {
584            sender.put_metric(
585                account_id,
586                region,
587                namespace,
588                metric_name,
589                value,
590                unit,
591                dimensions,
592                timestamp_ms,
593            );
594        }
595    }
596
597    pub fn with_cloudwatch_logs(mut self, sender: Arc<dyn CloudwatchLogsDelivery>) -> Self {
598        self.cloudwatch_logs = Some(sender);
599        self
600    }
601
602    /// Put log events to a CloudWatch Logs log group / stream.
603    /// Silently no-ops when no CloudWatch Logs sender is wired.
604    pub fn put_log_events(
605        &self,
606        account_id: &str,
607        log_group_name: &str,
608        log_stream_name: &str,
609        events: &[(i64, String)],
610    ) {
611        if let Some(ref sender) = self.cloudwatch_logs {
612            sender.put_log_events(account_id, log_group_name, log_stream_name, events);
613        }
614    }
615
616    pub fn with_ses_dispatcher(mut self, dispatcher: Arc<dyn SesSendEmailDispatcher>) -> Self {
617        self.ses_dispatcher = Some(dispatcher);
618        self
619    }
620
621    pub fn with_ecs_task_runner(mut self, runner: Arc<dyn EcsTaskRunner>) -> Self {
622        self.ecs_task_runner = Some(runner);
623        self
624    }
625
626    pub fn with_elbv2_target_registration(mut self, reg: Arc<dyn Elbv2TargetRegistration>) -> Self {
627        self.elbv2_target_registration = Some(reg);
628        self
629    }
630
631    /// Register targets with an ELBv2 target group. Silently no-ops when
632    /// no ELBv2 target registration hook is wired.
633    pub fn register_elbv2_targets(
634        &self,
635        account_id: &str,
636        target_group_arn: &str,
637        targets: Vec<(String, Option<i64>)>,
638    ) {
639        if let Some(ref reg) = self.elbv2_target_registration {
640            reg.register_targets(account_id, target_group_arn, targets);
641        }
642    }
643
644    /// Deregister targets from an ELBv2 target group. Silently no-ops
645    /// when no ELBv2 target registration hook is wired.
646    pub fn deregister_elbv2_targets(
647        &self,
648        account_id: &str,
649        target_group_arn: &str,
650        targets: Vec<(String, Option<i64>)>,
651    ) {
652        if let Some(ref reg) = self.elbv2_target_registration {
653            reg.deregister_targets(account_id, target_group_arn, targets);
654        }
655    }
656
657    /// Send an email via SES. Returns `Err` when no SES dispatcher is
658    /// wired or the underlying impl rejects (bad source/dest).
659    #[allow(clippy::too_many_arguments)]
660    pub fn send_ses_email(
661        &self,
662        account_id: &str,
663        from: &str,
664        to: Vec<String>,
665        cc: Vec<String>,
666        bcc: Vec<String>,
667        subject: Option<&str>,
668        text_body: Option<&str>,
669        html_body: Option<&str>,
670    ) -> Result<(), String> {
671        match self.ses_dispatcher {
672            Some(ref d) => {
673                d.send_email(account_id, from, to, cc, bcc, subject, text_body, html_body)
674            }
675            None => Err("SES dispatcher not configured".to_string()),
676        }
677    }
678
679    /// Run an ECS task. Returns `Err` when no ECS runner is wired or
680    /// the impl rejects (unknown cluster / task definition).
681    pub fn run_ecs_task(
682        &self,
683        account_id: &str,
684        cluster: &str,
685        task_definition: &str,
686        launch_type: Option<&str>,
687        count: usize,
688    ) -> Result<(), String> {
689        match self.ecs_task_runner {
690            Some(ref r) => r.run_task(account_id, cluster, task_definition, launch_type, count),
691            None => Err("ECS task runner not configured".to_string()),
692        }
693    }
694
695    pub fn with_s3(mut self, sender: Arc<dyn S3Delivery>) -> Self {
696        self.s3_writer = Some(sender);
697        self
698    }
699
700    pub fn with_firehose(mut self, sender: Arc<dyn FirehoseDelivery>) -> Self {
701        self.firehose_sender = Some(sender);
702        self
703    }
704
705    /// Send a single record to a Firehose delivery stream by ARN.
706    /// Silently no-ops when no Firehose sender is wired (in-process
707    /// tests). Production wiring goes through fakecloud_firehose.
708    pub fn put_record_to_firehose(&self, delivery_stream_arn: &str, data: &[u8]) {
709        if let Some(ref sender) = self.firehose_sender {
710            sender.put_record(delivery_stream_arn, data);
711        }
712    }
713
714    /// Write content to S3. Returns Err when no S3 writer is wired or
715    /// when the underlying impl rejects the bucket / payload.
716    pub fn put_object_to_s3(
717        &self,
718        account_id: &str,
719        bucket: &str,
720        key: &str,
721        body: Vec<u8>,
722        content_type: Option<&str>,
723    ) -> Result<(), String> {
724        match self.s3_writer {
725            Some(ref sender) => sender.put_object(account_id, bucket, key, body, content_type),
726            None => Err("S3 writer not configured".to_string()),
727        }
728    }
729
730    /// Read content from S3. Returns Err when no S3 client is wired or
731    /// when the underlying impl cannot resolve the object.
732    pub fn get_object_from_s3(
733        &self,
734        account_id: &str,
735        bucket: &str,
736        key: &str,
737    ) -> Result<Vec<u8>, String> {
738        match self.s3_writer {
739            Some(ref sender) => sender.get_object(account_id, bucket, key),
740            None => Err("S3 client not configured".to_string()),
741        }
742    }
743
744    pub fn with_sqs(mut self, sender: Arc<dyn SqsDelivery>) -> Self {
745        self.sqs_sender = Some(sender);
746        self
747    }
748
749    pub fn with_sns(mut self, sender: Arc<dyn SnsDelivery>) -> Self {
750        self.sns_sender = Some(sender);
751        self
752    }
753
754    pub fn with_eventbridge(mut self, sender: Arc<dyn EventBridgeDelivery>) -> Self {
755        self.eventbridge_sender = Some(sender);
756        self
757    }
758
759    pub fn with_lambda(mut self, invoker: Arc<dyn LambdaDelivery>) -> Self {
760        self.lambda_invoker = Some(invoker);
761        self
762    }
763
764    pub fn with_kinesis(mut self, sender: Arc<dyn KinesisDelivery>) -> Self {
765        self.kinesis_sender = Some(sender);
766        self
767    }
768
769    /// Put a record to a Kinesis Data Stream identified by ARN.
770    /// Silently no-ops when no Kinesis sender is wired.
771    pub fn put_record_to_kinesis(&self, stream_arn: &str, data: &str, partition_key: &str) {
772        if let Some(ref sender) = self.kinesis_sender {
773            sender.put_record(stream_arn, data, partition_key);
774        }
775    }
776
777    pub fn with_sagemaker_pipeline(mut self, starter: Arc<dyn SageMakerPipelineDelivery>) -> Self {
778        self.sagemaker_pipeline_starter = Some(starter);
779        self
780    }
781
782    /// Start a SageMaker pipeline execution if a starter is wired.
783    pub fn start_sagemaker_pipeline(&self, pipeline_arn: &str, parameters: &serde_json::Value) {
784        if let Some(ref starter) = self.sagemaker_pipeline_starter {
785            starter.start_pipeline_execution(pipeline_arn, parameters);
786        }
787    }
788
789    pub fn with_stepfunctions(mut self, starter: Arc<dyn StepFunctionsDelivery>) -> Self {
790        self.stepfunctions_starter = Some(starter);
791        self
792    }
793
794    /// The stored ARN of the SQS queue a QueueUrl names, if it exists.
795    pub fn sqs_queue_arn_for_url(&self, queue_url: &str) -> Option<String> {
796        self.sqs_sender.as_ref()?.queue_arn_for_url(queue_url)
797    }
798
799    /// Send a message to an SQS queue identified by ARN.
800    pub fn send_to_sqs(
801        &self,
802        queue_arn: &str,
803        message_body: &str,
804        attributes: &HashMap<String, String>,
805    ) {
806        if let Some(ref sender) = self.sqs_sender {
807            sender.deliver_to_queue(queue_arn, message_body, attributes);
808        }
809    }
810
811    /// Send a message to an SQS queue with message attributes and FIFO fields.
812    pub fn send_to_sqs_with_attrs(
813        &self,
814        queue_arn: &str,
815        message_body: &str,
816        message_attributes: &HashMap<String, SqsMessageAttribute>,
817        message_group_id: Option<&str>,
818        message_dedup_id: Option<&str>,
819    ) {
820        if let Some(ref sender) = self.sqs_sender {
821            sender.deliver_to_queue_with_attrs(
822                queue_arn,
823                message_body,
824                message_attributes,
825                message_group_id,
826                message_dedup_id,
827            );
828        }
829    }
830
831    /// Fallible SQS send — returns `Err` when the target queue does not
832    /// exist, so callers (Scheduler) can route to a DLQ. Returns
833    /// `Err(QueueNotFound)` when no SQS sender is wired up at all,
834    /// matching the "target unreachable" semantics Scheduler relies on.
835    pub fn try_send_to_sqs_with_attrs(
836        &self,
837        queue_arn: &str,
838        message_body: &str,
839        message_attributes: &HashMap<String, SqsMessageAttribute>,
840        message_group_id: Option<&str>,
841        message_dedup_id: Option<&str>,
842    ) -> Result<(), SqsDeliveryError> {
843        match self.sqs_sender {
844            Some(ref sender) => sender.try_deliver_to_queue_with_attrs(
845                queue_arn,
846                message_body,
847                message_attributes,
848                message_group_id,
849                message_dedup_id,
850            ),
851            None => Err(SqsDeliveryError::QueueNotFound(queue_arn.to_string())),
852        }
853    }
854
855    /// Publish a message to an SNS topic identified by ARN.
856    pub fn publish_to_sns(&self, topic_arn: &str, message: &str, subject: Option<&str>) {
857        if let Some(ref sender) = self.sns_sender {
858            sender.publish_to_topic(topic_arn, message, subject);
859        }
860    }
861
862    /// Put an event onto an EventBridge bus in the default account.
863    pub fn put_event_to_eventbridge(
864        &self,
865        source: &str,
866        detail_type: &str,
867        detail: &str,
868        event_bus_name: &str,
869    ) {
870        if let Some(ref sender) = self.eventbridge_sender {
871            sender.put_event(source, detail_type, detail, event_bus_name);
872        }
873    }
874
875    /// Put an event onto an EventBridge bus in a specific account. Used by
876    /// Scheduler to deliver to cross-account event buses.
877    pub fn put_event_to_eventbridge_for_account(
878        &self,
879        source: &str,
880        detail_type: &str,
881        detail: &str,
882        event_bus_name: &str,
883        target_account_id: &str,
884    ) {
885        if let Some(ref sender) = self.eventbridge_sender {
886            sender.put_event_to_account(
887                source,
888                detail_type,
889                detail,
890                event_bus_name,
891                target_account_id,
892            );
893        }
894    }
895
896    /// Invoke a Lambda function identified by ARN.
897    pub async fn invoke_lambda(
898        &self,
899        function_arn: &str,
900        payload: &str,
901    ) -> Option<Result<Vec<u8>, String>> {
902        if let Some(ref invoker) = self.lambda_invoker {
903            Some(invoker.invoke_lambda(function_arn, payload).await)
904        } else {
905            None
906        }
907    }
908
909    /// Put a record to a Kinesis stream identified by ARN.
910    pub fn send_to_kinesis(&self, stream_arn: &str, data: &str, partition_key: &str) {
911        if let Some(ref sender) = self.kinesis_sender {
912            sender.put_record(stream_arn, data, partition_key);
913        }
914    }
915
916    /// Start a Step Functions execution identified by state machine ARN.
917    pub fn start_stepfunctions_execution(&self, state_machine_arn: &str, input: &str) {
918        if let Some(ref starter) = self.stepfunctions_starter {
919            starter.start_execution(state_machine_arn, input);
920        }
921    }
922}
923
924impl Default for DeliveryBus {
925    fn default() -> Self {
926        Self::new()
927    }
928}
929
930#[cfg(test)]
931mod tests {
932    use super::*;
933    use std::sync::atomic::{AtomicUsize, Ordering};
934    use std::sync::Arc;
935
936    // Mock implementations
937    struct MockSqs {
938        call_count: AtomicUsize,
939    }
940    impl SqsDelivery for MockSqs {
941        fn deliver_to_queue(
942            &self,
943            _queue_arn: &str,
944            _message_body: &str,
945            _attributes: &HashMap<String, String>,
946        ) {
947            self.call_count.fetch_add(1, Ordering::SeqCst);
948        }
949    }
950
951    struct MockSns {
952        call_count: AtomicUsize,
953    }
954    impl SnsDelivery for MockSns {
955        fn publish_to_topic(&self, _topic_arn: &str, _message: &str, _subject: Option<&str>) {
956            self.call_count.fetch_add(1, Ordering::SeqCst);
957        }
958    }
959
960    struct MockEventBridge {
961        call_count: AtomicUsize,
962    }
963    impl EventBridgeDelivery for MockEventBridge {
964        fn put_event(
965            &self,
966            _source: &str,
967            _detail_type: &str,
968            _detail: &str,
969            _event_bus_name: &str,
970        ) {
971            self.call_count.fetch_add(1, Ordering::SeqCst);
972        }
973    }
974
975    struct MockKinesis {
976        call_count: AtomicUsize,
977    }
978    impl KinesisDelivery for MockKinesis {
979        fn put_record(&self, _stream_arn: &str, _data: &str, _partition_key: &str) {
980            self.call_count.fetch_add(1, Ordering::SeqCst);
981        }
982    }
983
984    struct MockStepFunctions {
985        call_count: AtomicUsize,
986    }
987    impl StepFunctionsDelivery for MockStepFunctions {
988        fn start_execution(&self, _state_machine_arn: &str, _input: &str) {
989            self.call_count.fetch_add(1, Ordering::SeqCst);
990        }
991    }
992
993    #[test]
994    fn delivery_bus_new_has_no_senders() {
995        let bus = DeliveryBus::new();
996        // Calling methods without senders should be no-ops
997        bus.send_to_sqs("arn:queue", "body", &HashMap::new());
998        bus.publish_to_sns("arn:topic", "msg", None);
999        bus.put_event_to_eventbridge("src", "type", "{}", "default");
1000        bus.send_to_kinesis("arn:stream", "data", "pk");
1001        bus.start_stepfunctions_execution("arn:sfn", "{}");
1002        // No panics = success
1003    }
1004
1005    #[test]
1006    fn delivery_bus_default_is_same_as_new() {
1007        let bus = DeliveryBus::default();
1008        bus.send_to_sqs("arn:q", "b", &HashMap::new());
1009    }
1010
1011    #[test]
1012    fn send_to_sqs_calls_sender() {
1013        let mock = Arc::new(MockSqs {
1014            call_count: AtomicUsize::new(0),
1015        });
1016        let bus = DeliveryBus::new().with_sqs(mock.clone());
1017
1018        bus.send_to_sqs("arn:queue", "msg", &HashMap::new());
1019        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1020
1021        bus.send_to_sqs("arn:queue2", "msg2", &HashMap::new());
1022        assert_eq!(mock.call_count.load(Ordering::SeqCst), 2);
1023    }
1024
1025    #[test]
1026    fn send_to_sqs_with_attrs_calls_sender() {
1027        let mock = Arc::new(MockSqs {
1028            call_count: AtomicUsize::new(0),
1029        });
1030        let bus = DeliveryBus::new().with_sqs(mock.clone());
1031
1032        let mut attrs = HashMap::new();
1033        attrs.insert(
1034            "key".to_string(),
1035            SqsMessageAttribute {
1036                data_type: "String".to_string(),
1037                string_value: Some("val".to_string()),
1038                binary_value: None,
1039            },
1040        );
1041        bus.send_to_sqs_with_attrs("arn:q", "body", &attrs, Some("group"), Some("dedup"));
1042        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1043    }
1044
1045    #[test]
1046    fn publish_to_sns_calls_sender() {
1047        let mock = Arc::new(MockSns {
1048            call_count: AtomicUsize::new(0),
1049        });
1050        let bus = DeliveryBus::new().with_sns(mock.clone());
1051
1052        bus.publish_to_sns("arn:topic", "message", Some("subject"));
1053        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1054    }
1055
1056    #[test]
1057    fn put_event_to_eventbridge_calls_sender() {
1058        let mock = Arc::new(MockEventBridge {
1059            call_count: AtomicUsize::new(0),
1060        });
1061        let bus = DeliveryBus::new().with_eventbridge(mock.clone());
1062
1063        bus.put_event_to_eventbridge("aws.s3", "Object Created", "{}", "default");
1064        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1065    }
1066
1067    #[test]
1068    fn send_to_kinesis_calls_sender() {
1069        let mock = Arc::new(MockKinesis {
1070            call_count: AtomicUsize::new(0),
1071        });
1072        let bus = DeliveryBus::new().with_kinesis(mock.clone());
1073
1074        bus.send_to_kinesis("arn:stream", "data", "partition-key");
1075        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1076    }
1077
1078    #[test]
1079    fn start_stepfunctions_calls_sender() {
1080        let mock = Arc::new(MockStepFunctions {
1081            call_count: AtomicUsize::new(0),
1082        });
1083        let bus = DeliveryBus::new().with_stepfunctions(mock.clone());
1084
1085        bus.start_stepfunctions_execution("arn:sfn:machine", r#"{"key":"val"}"#);
1086        assert_eq!(mock.call_count.load(Ordering::SeqCst), 1);
1087    }
1088
1089    #[test]
1090    fn builder_chaining_works() {
1091        let sqs = Arc::new(MockSqs {
1092            call_count: AtomicUsize::new(0),
1093        });
1094        let sns = Arc::new(MockSns {
1095            call_count: AtomicUsize::new(0),
1096        });
1097        let eb = Arc::new(MockEventBridge {
1098            call_count: AtomicUsize::new(0),
1099        });
1100        let kin = Arc::new(MockKinesis {
1101            call_count: AtomicUsize::new(0),
1102        });
1103        let sfn = Arc::new(MockStepFunctions {
1104            call_count: AtomicUsize::new(0),
1105        });
1106
1107        let bus = DeliveryBus::new()
1108            .with_sqs(sqs.clone())
1109            .with_sns(sns.clone())
1110            .with_eventbridge(eb.clone())
1111            .with_kinesis(kin.clone())
1112            .with_stepfunctions(sfn.clone());
1113
1114        bus.send_to_sqs("q", "m", &HashMap::new());
1115        bus.publish_to_sns("t", "m", None);
1116        bus.put_event_to_eventbridge("s", "d", "{}", "b");
1117        bus.send_to_kinesis("s", "d", "k");
1118        bus.start_stepfunctions_execution("sm", "{}");
1119
1120        assert_eq!(sqs.call_count.load(Ordering::SeqCst), 1);
1121        assert_eq!(sns.call_count.load(Ordering::SeqCst), 1);
1122        assert_eq!(eb.call_count.load(Ordering::SeqCst), 1);
1123        assert_eq!(kin.call_count.load(Ordering::SeqCst), 1);
1124        assert_eq!(sfn.call_count.load(Ordering::SeqCst), 1);
1125    }
1126
1127    #[tokio::test]
1128    async fn invoke_lambda_returns_none_without_invoker() {
1129        let bus = DeliveryBus::new();
1130        let result = bus.invoke_lambda("arn:lambda", "{}").await;
1131        assert!(result.is_none());
1132    }
1133}
fakecloud_core/delivery.rs

fakecloud_core/
delivery.rs
