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

fakecloud_core/
delivery.rs
