feat(transform): Add Send to AWS EventBridge (#203)

* feat(transform): Add Send to AWS EventBridge Transform

* refactor: AWS SDK v2

* build(terraform): Add Bus Support to AWS EventBridge

* docs(examples): Add EventBridge Lambda Bus Example

* style(terraform): Formatting

* docs(Transform): Update Comments

* docs: Updates for Future Breaking Changes

CONTRIBUTING.md

@@ -99,6 +99,10 @@ Substation commonly uses these io compatible containers:
 
 ### Naming Conventions
 
+#### Breaking Changes
+
+Any change that modifies the public API of Go packages and applications is a breaking change, and any source code that has non-obvious impact on the public API should be tagged with `BREAKING CHANGE` in a comment.
+
 #### Errors
 
 Errors should always start with `err` (or `Err`, if they are public). Commonly used errors are defined in [internal/errors.go](internal/errors.go).

build/config/substation.libsonnet

@@ -791,6 +791,25 @@
           type: type,
           settings: std.prune(std.mergePatch(default, $.helpers.abbv(s))),
         },
+        eventbridge(settings={}): {
+          local type = 'send_aws_eventbridge',
+          local default = {
+            id: $.helpers.id(type, settings),
+            batch: $.config.batch,
+            auxiliary_transforms: null,
+            aws: $.config.aws,
+            retry: $.config.retry,
+            arn: null,
+            description: null,
+          },
+          local s = std.mergePatch(settings, {
+            auxiliary_transforms: if std.objectHas(settings, 'auxiliary_transforms') then settings.auxiliary_transforms else if std.objectHas(settings, 'aux_tforms') then settings.aux_tforms else null,
+            aux_tforms: null,
+          }),
+
+          type: type,
+          settings: std.prune(std.mergePatch(default, $.helpers.abbv(s))),
+        },
         firehose(settings={}): $.transform.send.aws.kinesis_data_firehose(settings=settings),
         kinesis_data_firehose(settings={}): {
           local type = 'send_aws_kinesis_data_firehose',

build/terraform/aws/eventbridge/lambda/README.md

@@ -0,0 +1,44 @@
+<!-- BEGIN_TF_DOCS -->
+## Requirements
+
+| Name | Version |
+|------|---------|
+| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | ~> 1.2 |
+| <a name="requirement_aws"></a> [aws](#requirement\_aws) | ~> 5.0 |
+
+## Providers
+
+| Name | Version |
+|------|---------|
+| <a name="provider_aws"></a> [aws](#provider\_aws) | ~> 5.0 |
+| <a name="provider_random"></a> [random](#provider\_random) | n/a |
+
+## Modules
+
+No modules.
+
+## Resources
+
+| Name | Type |
+|------|------|
+| [aws_cloudwatch_event_rule.rule](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/cloudwatch_event_rule) | resource |
+| [aws_cloudwatch_event_target.target](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/cloudwatch_event_target) | resource |
+| [aws_iam_policy.access](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_policy) | resource |
+| [aws_iam_role_policy_attachment.access](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |
+| [aws_lambda_permission.allow_cloudwatch](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_permission) | resource |
+| [random_uuid.id](https://registry.terraform.io/providers/hashicorp/random/latest/docs/resources/uuid) | resource |
+| [aws_caller_identity.current](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/caller_identity) | data source |
+| [aws_iam_policy_document.access](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/iam_policy_document) | data source |
+
+## Inputs
+
+| Name | Description | Type | Default | Required |
+|------|-------------|------|---------|:--------:|
+| <a name="input_access"></a> [access](#input\_access) | List of IAM ARNs that are granted access to the resource. | `list(string)` | `[]` | no |
+| <a name="input_config"></a> [config](#input\_config) | Configuration for the EventBridge Lambda rule:<br><br>    * name:         The name of the rule.<br>    * description:  The description of the rule.<br>    * function:     The Lambda function to invoke when the rule is triggered.<br>    * event\_bus\_arn: The ARN of the event bus to associate with the rule. If not provided, the default event bus is used.<br>    * event\_pattern: The event pattern for the rule. If not provided, the rule is schedule-based.<br>    * schedule:     The schedule expression for the rule. If not provided, the rule is event-based. | <pre>object({<br>    name        = string<br>    description = string<br>    function = object({<br>      arn  = string<br>      name = string<br>    })<br>    <br>    # Optional<br>    event_bus_arn = optional(string, null)<br>    event_pattern = optional(string, null)<br>    schedule = optional(string, null)<br>  })</pre> | n/a | yes |
+| <a name="input_tags"></a> [tags](#input\_tags) | Tags to apply to all resources. | `map(any)` | `{}` | no |
+
+## Outputs
+
+No outputs.
+<!-- END_TF_DOCS -->

build/terraform/aws/eventbridge/lambda/_variables.tf

@@ -0,0 +1,37 @@
+variable "config" {
+  type = object({
+    name        = string
+    description = string
+    function = object({
+      arn  = string
+      name = string
+    })
+
+    # Optional
+    event_bus_arn = optional(string, null)
+    event_pattern = optional(string, null)
+    schedule      = optional(string, null)
+  })
+  description = <<EOH
+    Configuration for the EventBridge Lambda rule:
+
+    * name:         The name of the rule.
+    * description:  The description of the rule.
+    * function:     The Lambda function to invoke when the rule is triggered.
+    * event_bus_arn: The ARN of the event bus to associate with the rule. If not provided, the default event bus is used.
+    * event_pattern: The event pattern for the rule. If not provided, the rule is schedule-based.
+    * schedule:     The schedule expression for the rule. If not provided, the rule is event-based.
+EOH
+}
+
+variable "tags" {
+  type        = map(any)
+  default     = {}
+  description = "Tags to apply to all resources."
+}
+
+variable "access" {
+  type        = list(string)
+  default     = []
+  description = "List of IAM ARNs that are granted access to the resource."
+}

build/terraform/aws/eventbridge/lambda/main.tf

@@ -0,0 +1,68 @@
+data "aws_caller_identity" "current" {}
+
+resource "random_uuid" "id" {}
+
+resource "aws_cloudwatch_event_rule" "rule" {
+  name                = var.config.name
+  description         = var.config.description
+  event_bus_name      = var.config.event_bus_arn != null ? var.config.event_bus_arn : "default"
+  schedule_expression = var.config.schedule != null ? var.config.schedule : null
+  event_pattern       = var.config.event_pattern != null ? var.config.event_pattern : null
+  tags                = var.tags
+}
+
+resource "aws_cloudwatch_event_target" "target" {
+  rule      = aws_cloudwatch_event_rule.rule.name
+  target_id = var.config.name
+  arn       = var.config.function.arn
+}
+
+resource "aws_lambda_permission" "allow_cloudwatch" {
+  statement_id  = "AllowExecutionFromCloudWatch"
+  action        = "lambda:InvokeFunction"
+  function_name = var.config.function.name
+  principal     = "events.amazonaws.com"
+  source_arn    = aws_cloudwatch_event_rule.rule.arn
+}
+
+# Applies the policy to each role in the access list.
+resource "aws_iam_role_policy_attachment" "access" {
+  count      = length(var.access)
+  role       = var.access[count.index]
+  policy_arn = aws_iam_policy.access.arn
+}
+
+resource "aws_iam_policy" "access" {
+  name        = "substation-eventbridge-${resource.random_uuid.id.id}"
+  description = "Policy that grants access to the Substation ${var.config.name} EventBridge rule."
+  policy      = data.aws_iam_policy_document.access.json
+}
+
+data "aws_iam_policy_document" "access" {
+  # Always allow access to the default event bus for the account.
+  statement {
+    effect = "Allow"
+    actions = [
+      "events:PutEvents",
+    ]
+
+    resources = [
+      "arn:aws:events:*:${data.aws_caller_identity.current.account_id}:event-bus/default",
+    ]
+  }
+
+  dynamic "statement" {
+    for_each = var.config.event_bus_arn != null ? [1] : []
+
+    content {
+      effect = "Allow"
+      actions = [
+        "events:PutEvents",
+      ]
+
+      resources = [
+        var.config.event_bus_arn,
+      ]
+    }
+  }
+}

examples/terraform/aws/README.md

@@ -139,6 +139,38 @@ flowchart LR
     mdEnrichmentTransforms --- dynamodb
 ```
 
+# EventBridge
+
+## Lambda Bus
+
+Deploys a data pipeline that sends data from an EventBridge event bus to a Lambda function.
+
+```mermaid
+flowchart LR
+    %% resources
+    ebb([EventBridge Bus])
+    ebs([EventBridge Scheduler])
+
+    producerHandler[[Handler]]
+    producerTransforms[Transforms]
+
+    consumerHandler[[Handler]]
+    consumerTransforms[Transforms]
+
+    %% connections
+    ebs --> ebs
+    ebs --> producerHandler
+    subgraph Substation Producer Node 
+    producerHandler --> producerTransforms
+    end
+
+    producerTransforms --> ebb --> consumerHandler
+
+    subgraph Substation Consumer Node 
+    consumerHandler  --> consumerTransforms
+    end
+```
+
 # Firehose
 
 ## Data Transform

examples/terraform/aws/eventbridge/lambda_bus/config/consumer/config.jsonnet

@@ -0,0 +1,8 @@
+local sub = import '../../../../../../../build/config/substation.libsonnet';
+
+{
+  concurrency: 1,
+  transforms: [
+    sub.tf.send.stdout(),
+  ],
+}

examples/terraform/aws/eventbridge/lambda_bus/config/producer/config.jsonnet

@@ -0,0 +1,11 @@
+local sub = import '../../../../../../../build/config/substation.libsonnet';
+
+{
+  concurrency: 1,
+  transforms: [
+    sub.tf.time.now({object: {target_key: 'ts'}}),
+    sub.tf.obj.insert({object: {target_key: 'message'}, value: 'Hello from the EventBridge scheduler!'}),
+    // This sends the event to the default bus.
+    sub.tf.send.aws.eventbridge(),
+  ],
+}

examples/terraform/aws/eventbridge/lambda_bus/terraform/_resources.tf

@@ -0,0 +1,17 @@
+module "appconfig" {
+  source = "../../../../../../build/terraform/aws/appconfig"
+
+  config = {
+    name         = "substation"
+    environments = [{ name = "example" }]
+  }
+}
+
+module "ecr" {
+  source = "../../../../../../build/terraform/aws/ecr"
+
+  config = {
+    name         = "substation"
+    force_delete = true
+  }
+}

examples/terraform/aws/eventbridge/lambda_bus/terraform/consumer.tf

@@ -0,0 +1,35 @@
+module "eventbridge_consumer" {
+  source = "../../../../../../build/terraform/aws/eventbridge/lambda"
+
+  config = {
+    name        = "substation_consumer"
+    description = "Routes messages from any Substation producer to a Substation Lambda consumer."
+    function    = module.lambda_consumer # This is the Lambda function that will be invoked.
+    event_pattern = jsonencode({
+      # This matches every event sent by any Substation app.
+      source = [{ "wildcard" : "substation.*" }]
+    })
+  }
+
+  access = [
+    module.lambda_producer.role.name,
+  ]
+}
+
+module "lambda_consumer" {
+  source    = "../../../../../../build/terraform/aws/lambda"
+  appconfig = module.appconfig
+
+  config = {
+    name        = "consumer"
+    description = "Substation node that is invoked by the EventBridge bus."
+    image_uri   = "${module.ecr.url}:v1.5.0"
+    image_arm   = true
+
+    env = {
+      "SUBSTATION_CONFIG" : "http://localhost:2772/applications/substation/environments/example/configurations/consumer"
+      "SUBSTATION_LAMBDA_HANDLER" : "AWS_LAMBDA"
+      "SUBSTATION_DEBUG" : true
+    }
+  }
+}