Add EventBridge Lambda target delivery (#146)

* Add EventBridge Lambda target delivery

- Routes EventBridge Lambda targets through the Go Lambda invocation path with CloudWatch Logs recording.

- Keeps local Node.js handler execution behind the existing signed localhost allow-local gate.

- Adds Go and AWS SDK coverage plus docs for EventBridge-to-Lambda delivery.

* Update AWS skill EventBridge Lambda docs

* Document EventBridge Lambda known key gate

* Format AWS EventBridge e2e test

Author: ctate

README.md

@@ -778,12 +778,12 @@ In the native Go runtime, `@aws-sdk/client-sns` v3 can use the `/sns/` endpoint
 - `AddPermission`, `RemovePermission`
 
 ### EventBridge
-In the native Go runtime, `@aws-sdk/client-eventbridge` v3 can use the `/events/` endpoint directly. The SDK sends `X-Amz-Target: AWSEvents.<Action>` JSON requests and receives JSON responses. Matching events can be delivered to SQS queues and SNS topics.
+In the native Go runtime, `@aws-sdk/client-eventbridge` v3 can use the `/events/` endpoint directly. The SDK sends `X-Amz-Target: AWSEvents.<Action>` JSON requests and receives JSON responses. Matching events can be delivered to SQS queues, SNS topics, and Lambda functions. Lambda targets create CloudWatch Logs entries; zipped Node.js handlers run only when `npx emulate` is started with `--allow-local-lambda` and the EventBridge request uses a direct localhost endpoint signed by a known AWS access key.
 
 - `CreateEventBus`, `DeleteEventBus`, `ListEventBuses`
 - `PutRule`, `DescribeRule`, `ListRules`, `DeleteRule`, `EnableRule`, `DisableRule`
-- `PutTargets`, `ListTargetsByRule`, `RemoveTargets`
-- `PutEvents` with rule pattern matching
+- `PutTargets`, `ListTargetsByRule`, `RemoveTargets` for SQS, SNS, and Lambda targets
+- `PutEvents` with rule pattern matching and target delivery
 - `TagResource`, `UntagResource`, `ListTagsForResource`
 
 ### DynamoDB

apps/web/app/docs/aws/page.mdx

@@ -60,12 +60,12 @@ In the native Go runtime, `@aws-sdk/client-sns` v3 can use the `/sns/` endpoint
 
 ## EventBridge
 
-In the native Go runtime, `@aws-sdk/client-eventbridge` v3 can use the `/events/` endpoint directly. The SDK sends `X-Amz-Target: AWSEvents.<Action>` JSON requests and receives JSON responses. Matching events can be delivered to SQS queues and SNS topics.
+In the native Go runtime, `@aws-sdk/client-eventbridge` v3 can use the `/events/` endpoint directly. The SDK sends `X-Amz-Target: AWSEvents.<Action>` JSON requests and receives JSON responses. Matching events can be delivered to SQS queues, SNS topics, and Lambda functions. Lambda targets create CloudWatch Logs entries; zipped Node.js handlers run only when `npx emulate` is started with `--allow-local-lambda` and the EventBridge request uses a direct localhost endpoint signed by a known AWS access key.
 
 - `CreateEventBus` / `DeleteEventBus` / `ListEventBuses` - event bus lifecycle and discovery
 - `PutRule` / `DescribeRule` / `ListRules` / `DeleteRule` / `EnableRule` / `DisableRule` - rule lifecycle and state
-- `PutTargets` / `ListTargetsByRule` / `RemoveTargets` - SQS and SNS target management
-- `PutEvents` - publish events with rule pattern matching
+- `PutTargets` / `ListTargetsByRule` / `RemoveTargets` - SQS, SNS, and Lambda target management
+- `PutEvents` - publish events with rule pattern matching and target delivery
 - `TagResource` / `UntagResource` / `ListTagsForResource` - event bus and rule tags
 
 ## DynamoDB

internal/services/aws/eventbridge/handler.go

@@ -16,6 +16,7 @@ import (
 
 	corestore "github.com/vercel-labs/emulate/internal/core/store"
 	"github.com/vercel-labs/emulate/internal/services/aws/gateway"
+	awslambda "github.com/vercel-labs/emulate/internal/services/aws/lambda"
 	"github.com/vercel-labs/emulate/internal/services/aws/protocols"
 )
 
@@ -30,24 +31,31 @@ type eventBusRef struct {
 }
 
 type Handler struct {
-	EventBuses       *corestore.Collection
-	EventRules       *corestore.Collection
-	EventTargets     *corestore.Collection
-	EventDeliveries  *corestore.Collection
-	SQSQueues        *corestore.Collection
-	SQSMessages      *corestore.Collection
-	SNSTopics        *corestore.Collection
-	SNSSubscriptions *corestore.Collection
-	SNSDeliveries    *corestore.Collection
-	AccountID        string
-	Region           string
-	Now              func() time.Time
-	IDGenerator      func(string) string
+	EventBuses               *corestore.Collection
+	EventRules               *corestore.Collection
+	EventTargets             *corestore.Collection
+	EventDeliveries          *corestore.Collection
+	SQSQueues                *corestore.Collection
+	SQSMessages              *corestore.Collection
+	SNSTopics                *corestore.Collection
+	SNSSubscriptions         *corestore.Collection
+	SNSDeliveries            *corestore.Collection
+	LambdaFunctions          *corestore.Collection
+	LambdaVersions           *corestore.Collection
+	LambdaAliases            *corestore.Collection
+	LogGroups                *corestore.Collection
+	LogStreams               *corestore.Collection
+	LogEvents                *corestore.Collection
+	LambdaLocalCodeExecution bool
+	AccountID                string
+	Region                   string
+	Now                      func() time.Time
+	IDGenerator              func(string) string
 }
 
 var fallbackIDCounter atomic.Uint64
 
-func (h *Handler) Handle(_ *http.Request, ctx gateway.AwsRequestContext) protocols.ErrorResponse {
+func (h *Handler) Handle(req *http.Request, ctx gateway.AwsRequestContext) protocols.ErrorResponse {
 	requestID := ctx.RequestID
 	if requestID == "" {
 		requestID = h.generateID("req")
@@ -79,7 +87,7 @@ func (h *Handler) Handle(_ *http.Request, ctx gateway.AwsRequestContext) protoco
 	case "RemoveTargets":
 		response = h.removeTargets(ctx, requestID)
 	case "PutEvents":
-		response = h.putEvents(ctx, requestID)
+		response = h.putEvents(req, ctx, requestID)
 	case "TagResource":
 		response = h.tagResource(ctx, requestID)
 	case "UntagResource":
@@ -372,7 +380,7 @@ func (h *Handler) removeTargets(ctx gateway.AwsRequestContext, requestID string)
 	return jsonResponse(http.StatusOK, map[string]any{"FailedEntryCount": 0, "FailedEntries": []any{}})
 }
 
-func (h *Handler) putEvents(ctx gateway.AwsRequestContext, requestID string) protocols.ErrorResponse {
+func (h *Handler) putEvents(req *http.Request, ctx gateway.AwsRequestContext, requestID string) protocols.ErrorResponse {
 	entries := mapSlice(ctx.Input["Entries"])
 	if len(entries) == 0 {
 		return h.validation("Entries is required.", requestID)
@@ -421,7 +429,7 @@ func (h *Handler) putEvents(ctx gateway.AwsRequestContext, requestID string) pro
 			"resources":   stringSlice(entry["Resources"]),
 			"detail":      detail,
 		}
-		h.deliverEvent(ctx, busName, eventID, event)
+		h.deliverEvent(req, ctx, busName, eventID, event)
 		out = append(out, map[string]any{"EventId": eventID})
 	}
 	return jsonResponse(http.StatusOK, map[string]any{"FailedEntryCount": failed, "Entries": out})
@@ -457,7 +465,7 @@ func (h *Handler) listTagsForResource(ctx gateway.AwsRequestContext, requestID s
 	return jsonResponse(http.StatusOK, map[string]any{"Tags": recordList(record["tags"])})
 }
 
-func (h *Handler) deliverEvent(ctx gateway.AwsRequestContext, busName string, eventID string, event map[string]any) {
+func (h *Handler) deliverEvent(req *http.Request, ctx gateway.AwsRequestContext, busName string, eventID string, event map[string]any) {
 	for _, rule := range h.EventRules.FindBy("event_bus_name", busName) {
 		if !h.sameScope(ctx, rule) || stringField(rule, "state") == "DISABLED" || !matchesPattern(rule, event) {
 			continue
@@ -473,6 +481,8 @@ func (h *Handler) deliverEvent(ctx gateway.AwsRequestContext, busName string, ev
 				status = "DELIVERED"
 			case h.deliverToSNS(ctx, target, body):
 				status = "DELIVERED"
+			case h.deliverToLambda(req, ctx, target, body):
+				status = "DELIVERED"
 			}
 			h.EventDeliveries.Insert(corestore.Record{
 				"account_id":      h.accountID(ctx),
@@ -572,6 +582,27 @@ func (h *Handler) deliverToSNS(ctx gateway.AwsRequestContext, target corestore.R
 	return delivered
 }
 
+func (h *Handler) deliverToLambda(req *http.Request, ctx gateway.AwsRequestContext, target corestore.Record, body string) bool {
+	if h.LambdaFunctions == nil {
+		return false
+	}
+	lambdaHandler := awslambda.Handler{
+		Functions:               h.LambdaFunctions,
+		Versions:                h.LambdaVersions,
+		Aliases:                 h.LambdaAliases,
+		LogGroups:               h.LogGroups,
+		LogStreams:              h.LogStreams,
+		LogEvents:               h.LogEvents,
+		AccountID:               h.AccountID,
+		Region:                  h.Region,
+		AllowLocalCodeExecution: h.LambdaLocalCodeExecution,
+		Now:                     h.Now,
+		IDGenerator:             h.IDGenerator,
+	}
+	_, ok := lambdaHandler.InvokeFromEventSource(req, ctx, stringField(target, "arn"), []byte(body), h.generateID("req"), "EventBridge")
+	return ok
+}
+
 func (h *Handler) snsSQSBody(topic corestore.Record, subscription corestore.Record, messageID string, message string) string {
 	if strings.EqualFold(stringMapField(subscription, "attributes")["RawMessageDelivery"], "true") {
 		return message

internal/services/aws/lambda/handler.go

@@ -58,6 +58,13 @@ type functionIdentifier struct {
 	ARN       bool
 }
 
+type EventSourceInvokeResult struct {
+	ExecutedVersion string
+	FunctionError   string
+	LocalExecuted   bool
+	LogStreamName   string
+}
+
 var fallbackIDCounter atomic.Uint64
 
 func (h *Handler) Handle(req *http.Request, ctx gateway.AwsRequestContext) protocols.ErrorResponse {
@@ -419,6 +426,37 @@ func (h *Handler) invoke(req *http.Request, ctx gateway.AwsRequestContext, route
 	return lambdaBodyResponse(http.StatusOK, payload, headers)
 }
 
+func (h *Handler) InvokeFromEventSource(req *http.Request, ctx gateway.AwsRequestContext, targetARN string, payload []byte, requestID string, source string) (EventSourceInvokeResult, bool) {
+	fn, ok := h.findFunction(ctx, targetARN)
+	if !ok {
+		return EventSourceInvokeResult{}, false
+	}
+	parsed := parseFunctionIdentifier(targetARN)
+	invoked, executedVersion, _, ok := h.recordForQualifier(ctx, fn, parsed.Qualifier, requestID)
+	if !ok {
+		return EventSourceInvokeResult{}, false
+	}
+	if len(strings.TrimSpace(string(payload))) == 0 {
+		payload = []byte("{}")
+	}
+	logStreamName := h.lambdaLogStreamName(executedVersion)
+	source = strings.TrimSpace(source)
+	if source == "" {
+		source = "event source"
+	}
+	logLines := []string{"Lambda " + source + " invoke accepted RequestId: " + requestID}
+	result := EventSourceInvokeResult{ExecutedVersion: executedVersion, LogStreamName: logStreamName}
+	if h.localCodeExecutionAllowed(req, ctx) {
+		if local, ran := h.invokeLocalNode(ctx, invoked, executedVersion, invokedFunctionARN(fn, parsed.Qualifier), payload, requestID, logStreamName); ran {
+			logLines = local.Logs
+			result.FunctionError = local.FunctionError
+			result.LocalExecuted = true
+		}
+	}
+	h.recordInvocation(ctx, invoked, requestID, "Event", executedVersion, logStreamName, logLines)
+	return result, true
+}
+
 func (h *Handler) localCodeExecutionAllowed(req *http.Request, ctx gateway.AwsRequestContext) bool {
 	if !h.AllowLocalCodeExecution || ctx.Auth.Status != auth.StatusKnown || ctx.Auth.Credential == nil {
 		return false

internal/services/aws/service.go

@@ -154,17 +154,24 @@ func New(options Options) *Service {
 			Region:    defaultRegion,
 		},
 		events: awsevents.Handler{
-			EventBuses:       awsStore.EventBuses,
-			EventRules:       awsStore.EventRules,
-			EventTargets:     awsStore.EventTargets,
-			EventDeliveries:  awsStore.EventDeliveries,
-			SQSQueues:        awsStore.SQSQueues,
-			SQSMessages:      awsStore.SQSMessages,
-			SNSTopics:        awsStore.SNSTopics,
-			SNSSubscriptions: awsStore.SNSSubscriptions,
-			SNSDeliveries:    awsStore.SNSDeliveries,
-			AccountID:        defaultAccountID,
-			Region:           defaultRegion,
+			EventBuses:               awsStore.EventBuses,
+			EventRules:               awsStore.EventRules,
+			EventTargets:             awsStore.EventTargets,
+			EventDeliveries:          awsStore.EventDeliveries,
+			SQSQueues:                awsStore.SQSQueues,
+			SQSMessages:              awsStore.SQSMessages,
+			SNSTopics:                awsStore.SNSTopics,
+			SNSSubscriptions:         awsStore.SNSSubscriptions,
+			SNSDeliveries:            awsStore.SNSDeliveries,
+			LambdaFunctions:          awsStore.LambdaFunctions,
+			LambdaVersions:           awsStore.LambdaVersions,
+			LambdaAliases:            awsStore.LambdaAliases,
+			LogGroups:                awsStore.LogGroups,
+			LogStreams:               awsStore.LogStreams,
+			LogEvents:                awsStore.LogEvents,
+			LambdaLocalCodeExecution: options.LambdaLocalCodeExecution,
+			AccountID:                defaultAccountID,
+			Region:                   defaultRegion,
 		},
 		logs: awslogs.Handler{
 			LogGroups:  awsStore.LogGroups,

internal/services/aws/service_test.go

@@ -1711,6 +1711,128 @@ func TestServiceHandlesEventBridgeRuleAndSQSTarget(t *testing.T) {
 	}
 }
 
+func TestServiceHandlesEventBridgeLambdaTarget(t *testing.T) {
+	handler := newTestHandler()
+
+	create := executeAWSLambdaRequest(t, handler, http.MethodPost, "/2015-03-31/functions", map[string]any{
+		"FunctionName": "eventbridge-target",
+		"Runtime":      "python3.12",
+		"Role":         "arn:aws:iam::123456789012:role/lambda-execution-role",
+		"Handler":      "index.handler",
+		"Code":         map[string]any{"ZipFile": base64.StdEncoding.EncodeToString([]byte("stub"))},
+	})
+	if create.Code != http.StatusCreated {
+		t.Fatalf("create lambda status = %d, body = %s", create.Code, create.Body.String())
+	}
+	var created struct {
+		FunctionArn string `json:"FunctionArn"`
+	}
+	decodeJSONBody(t, create, &created)
+
+	res := executeAWSEventBridgeRequest(t, handler, "PutRule", map[string]any{
+		"Name":         "lambda-orders",
+		"EventPattern": `{"source":["app.orders"],"detail-type":["OrderCreated"]}`,
+	})
+	if res.Code != http.StatusOK {
+		t.Fatalf("put rule status = %d, body = %s", res.Code, res.Body.String())
+	}
+
+	res = executeAWSEventBridgeRequest(t, handler, "PutTargets", map[string]any{
+		"Rule": "lambda-orders",
+		"Targets": []map[string]any{{
+			"Id":  "lambda",
+			"Arn": created.FunctionArn,
+		}},
+	})
+	if res.Code != http.StatusOK || !strings.Contains(res.Body.String(), `"FailedEntryCount":0`) {
+		t.Fatalf("put targets status = %d, body = %s", res.Code, res.Body.String())
+	}
+
+	res = executeAWSEventBridgeRequest(t, handler, "PutEvents", map[string]any{
+		"Entries": []map[string]any{{"Source": "app.orders", "DetailType": "OrderCreated", "Detail": `{"id":"ord_1"}`}},
+	})
+	if res.Code != http.StatusOK || !strings.Contains(res.Body.String(), `"FailedEntryCount":0`) {
+		t.Fatalf("put events status = %d, body = %s", res.Code, res.Body.String())
+	}
+
+	logs := executeAWSLogsRequest(t, handler, "FilterLogEvents", map[string]any{"logGroupName": "/aws/lambda/eventbridge-target", "filterPattern": "EventBridge"})
+	if logs.Code != http.StatusOK {
+		t.Fatalf("filter logs status = %d, body = %s", logs.Code, logs.Body.String())
+	}
+	var logBody struct {
+		Events []struct {
+			Message string `json:"message"`
+		} `json:"events"`
+	}
+	decodeJSONBody(t, logs, &logBody)
+	if len(logBody.Events) == 0 || !strings.Contains(logBody.Events[0].Message, "Lambda EventBridge invoke accepted") {
+		t.Fatalf("unexpected lambda target logs: %#v", logBody.Events)
+	}
+}
+
+func TestServiceRunsEventBridgeLambdaTargetWithLocalNodeHandler(t *testing.T) {
+	if _, err := exec.LookPath("node"); err != nil {
+		t.Skip("node is required for local Lambda Node.js runner coverage")
+	}
+	handler := newTestHandlerWithOptions(Options{LambdaLocalCodeExecution: true})
+	const accessKeyID = "AKIAIOSFODNN7EXAMPLE"
+	zipFile := zipLambdaSource(t, map[string]string{"index.js": `exports.handler = async (event, context) => {
+  console.log("eventbridge lambda", event.detail.name, context.functionName);
+  return { ok: true };
+};
+`})
+
+	create := executeAWSLambdaRequest(t, handler, http.MethodPost, "/2015-03-31/functions", map[string]any{
+		"FunctionName": "eventbridge-local-target",
+		"Runtime":      "nodejs22.x",
+		"Role":         "arn:aws:iam::123456789012:role/lambda-execution-role",
+		"Handler":      "index.handler",
+		"Code":         map[string]any{"ZipFile": zipFile},
+	})
+	if create.Code != http.StatusCreated {
+		t.Fatalf("create lambda status = %d, body = %s", create.Code, create.Body.String())
+	}
+	var created struct {
+		FunctionArn string `json:"FunctionArn"`
+	}
+	decodeJSONBody(t, create, &created)
+
+	res := executeAWSEventBridgeRequestWithAccessKey(t, handler, "PutRule", map[string]any{
+		"Name":         "lambda-local-orders",
+		"EventPattern": `{"source":["app.orders"],"detail-type":["OrderCreated"]}`,
+	}, accessKeyID)
+	if res.Code != http.StatusOK {
+		t.Fatalf("put rule status = %d, body = %s", res.Code, res.Body.String())
+	}
+	res = executeAWSEventBridgeRequestWithAccessKey(t, handler, "PutTargets", map[string]any{
+		"Rule":    "lambda-local-orders",
+		"Targets": []map[string]any{{"Id": "lambda", "Arn": created.FunctionArn}},
+	}, accessKeyID)
+	if res.Code != http.StatusOK {
+		t.Fatalf("put targets status = %d, body = %s", res.Code, res.Body.String())
+	}
+	res = executeAWSEventBridgeRequestWithAccessKey(t, handler, "PutEvents", map[string]any{
+		"Entries": []map[string]any{{"Source": "app.orders", "DetailType": "OrderCreated", "Detail": `{"name":"Ada"}`}},
+	}, accessKeyID)
+	if res.Code != http.StatusOK {
+		t.Fatalf("put events status = %d, body = %s", res.Code, res.Body.String())
+	}
+
+	logs := executeAWSLogsRequest(t, handler, "FilterLogEvents", map[string]any{"logGroupName": "/aws/lambda/eventbridge-local-target", "filterPattern": "eventbridge lambda"})
+	if logs.Code != http.StatusOK {
+		t.Fatalf("filter logs status = %d, body = %s", logs.Code, logs.Body.String())
+	}
+	var logBody struct {
+		Events []struct {
+			Message string `json:"message"`
+		} `json:"events"`
+	}
+	decodeJSONBody(t, logs, &logBody)
+	if len(logBody.Events) != 1 || !strings.Contains(logBody.Events[0].Message, "eventbridge lambda Ada eventbridge-local-target") {
+		t.Fatalf("unexpected lambda target logs: %#v", logBody.Events)
+	}
+}
+
 func TestServiceHandlesEventBridgeCustomBusTagsAndSNSTarget(t *testing.T) {
 	handler := newTestHandler()
 
@@ -5301,6 +5423,7 @@ func executeAWSEventBridgeRequestWithAccessKey(t *testing.T, handler http.Handle
 		t.Fatal(err)
 	}
 	req := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/events/", bytes.NewReader(raw))
+	req.RemoteAddr = "127.0.0.1:1234"
 	req.Header.Set("Content-Type", "application/x-amz-json-1.1")
 	req.Header.Set("X-Amz-Target", "AWSEvents."+action)
 	req.Header.Set("X-Access-Key", accessKeyID)