feat(pipes): apply EventBridge Pipe enrichment before target delivery

[abanna]

Summary

PipesPoller delivered the raw source record straight to the target; pipe.getEnrichment() was parsed and stored but never invoked. This implements the AWS EventBridge Pipes source → filter → enrichment → target flow:

• When an enrichment Lambda is configured, invoke it once with the filtered events as a bare JSON array (matching AWS batch semantics) and forward its response to the target.
• A null/empty enrichment response skips the target (per AWS behavior).
• Resolve qualified Lambda ARNs (e.g. ...:function:NAME:$LATEST) when invoking the enrichment function from PipesTargetInvoker.

Without this, a pipe configured with an enrichment step silently bypassed it, delivering an unenriched payload to the target (Step Functions / SQS / etc.).

Type of change

• New feature (feat:)

AWS Compatibility

EventBridge Pipes enrichment: an enrichment resource (Lambda) is invoked between the filter and the target, receiving the filtered events as a JSON array and whose response becomes the target input; a null/empty response drops the event. Verified against the AWS Pipes documented behavior and exercised by PipesTargetInvokerTest.

Checklist

• ./mvnw test passes locally (PipesTargetInvokerTest: 24 tests, 0 failures — run in an eclipse-temurin:25-jdk container)
• New or updated integration test added (PipesTargetInvokerTest)
• Commit messages follow Conventional Commits

[greptile-apps]

Greptile Summary

This PR implements the AWS EventBridge Pipes source → filter → enrichment → target flow for SQS-sourced pipes. Previously, pipe.getEnrichment() was stored but never invoked; this wires it in for Lambda enrichments, with correct null-skip semantics, qualified ARN resolution, and FunctionError surfacing for both enrichment and target Lambda invocations.

• Enrichment path (SQS only): deliverEnrichedBatch invokes the enrichment Lambda once with the filtered batch as a bare JSON array, forwards its response to the target (raw for non-Lambda targets, array-wrapped for Lambda targets), and routes to the DLQ on failure. Null/empty enrichment responses skip the target while still consuming the source messages, matching AWS behavior.
• lambdaFunctionName utility: Correctly extracts the function name from qualified ARNs (e.g. …:function:NAME:$LATEST) by stripping the qualifier, fixing the previous lastIndexOf(':') approach that returned the qualifier instead.
• Known limitation documented: Kinesis, DynamoDB Streams, and Kafka sources bypass the enrichment step; this is called out in both code comments and docs/services/pipes.md as deferred work.

Confidence Score: 5/5

Safe to merge — the enrichment path is correctly scoped to the SQS source, its known limitations are explicitly documented, and all previously identified issues have been resolved.

All previously raised issues have been addressed: the qualified-ARN function name extraction is fixed, Lambda FunctionError is surfaced for both enrichment and target invocations, and the array-wrapping is correctly gated on isLambdaTarget. The known gap (enrichment bypassed for Kinesis/DynamoDB Streams/Kafka) is intentionally deferred and thoroughly documented in both code comments and the service docs. The new tests cover the key behavioral boundaries (null/empty skip, FunctionError throw, qualified ARN resolution, Lambda vs non-Lambda target payload shape). No new regressions are introduced.

No files require special attention.

Important Files Changed

Filename	Overview
src/main/java/io/github/hectorvent/floci/services/pipes/PipesPoller.java	Adds deliverEnrichedBatch for SQS-sourced enrichment; correctly gates Lambda-target array-wrapping via isLambdaTarget; documents the known Kinesis/DynamoDB/Kafka bypass in both code and docs.
src/main/java/io/github/hectorvent/floci/services/pipes/PipesTargetInvoker.java	Adds applyEnrichment with correct null-skip and FunctionError semantics; fixes qualified-ARN resolution via lambdaFunctionName; surfaces target Lambda FunctionError so the poller routes to DLQ instead of silently consuming the record.
src/test/java/io/github/hectorvent/floci/services/pipes/PipesTargetInvokerTest.java	Adds 9 tests covering Lambda invocation, qualified-ARN name extraction, null/empty/object/array skip semantics, unsupported enrichment type, FunctionError handling, and target Lambda FunctionError surfacing.
src/test/java/io/github/hectorvent/floci/services/pipes/PipesPollerTest.java	Adds unit tests for asEventArray and two integration-style tests verifying that non-Lambda targets receive the raw enrichment response while Lambda targets receive the array-wrapped form.
docs/services/pipes.md	Adds Enrichment section clearly documenting Lambda-only support, null-skip semantics, FunctionError DLQ routing, and the known SQS-only limitation for enrichment.

_{Reviews (7): Last reviewed commit: "docs: regenerate action tables" | Re-trigger Greptile}

[abanna]

Applied the Greptile P1: deliverEnrichedBatch no longer array-wraps the enrichment response for every target. The asEventArray wrapping is now gated on isLambdaTarget — a Lambda target still gets the SQSRecord[]-style batch, while a Step Functions / SQS / SNS / EventBridge target receives the raw enrichment response (array-wrapping those corrupted their input). This matches the non-enrichment delivery path's Lambda-vs-other handling. Added pollSqs tests for both a non-Lambda (raw) and a Lambda (wrapped) target.

On the summary's other note — enrichment is still only wired on the SQS poll path; the Kinesis/DynamoDB-Streams/Kafka paths don't yet run enrichment. That's a deliberate scope boundary for this PR (SQS is the common Pipes source); I'll track extending enrichment to the stream sources as a follow-up rather than widen this change.

Branch is up to date with main (merge) and green (35 tests).

[hectorvent]

Thanks @abanna, good catch on the enrichment step being parsed but never invoked, and the qualified ARN fix plus the FunctionError to DLQ change are real improvements on their own. Checked the behavior against the AWS Pipes enrichment docs.

What lines up with AWS: synchronous REQUEST_RESPONSE invocation, forwarding the enrichment response directly to the target ("EventBridge Pipes passes the enrichment responses directly to the configured target"), keeping array responses unchanged, and consuming the source batch when the enrichment filters everything out.

Two things worth addressing:

1. The empty response check is narrower than AWS. The docs say to skip the target the enrichment can return "", {}, or [], and that [{}] is the explicit way to invoke the target with an empty payload. Right now {} is forwarded, and for a Lambda target it gets wrapped into [{}], which is exactly the opposite of what AWS does:

// enrichment returns {}          AWS: skip the target
// this PR: forwards it, and for a Lambda target delivers [{}] (explicit invoke with empty payload)
if (resp.isEmpty() || "null".equals(resp)) { ... }   // should also treat {} and [] as empty

2. The enrichment branch is only wired into pollSqs. A Kinesis, DynamoDB Streams, or Kafka pipe with an enrichment configured still silently bypasses it, which is the same bug this PR fixes. Wiring the same check into the other pollers would be ideal, otherwise an explicit SQS only note in the PR and docs works too.

Smaller note: for the unsupported enrichment types (API destinations, API Gateway, SFN Express are all valid on AWS) the code warns and delivers the unenriched payload anyway. Failing the batch to the DLQ might be safer than silently delivering data the pipe was supposed to enrich.

Nice work overall, the test coverage on the batch shaping is thorough.

[abanna]

Thanks — addressed the enrichment points and merged current main in.

1. Empty-response semantics now match AWS. applyEnrichment skips the target for "", null, {}, and [] (parsed, so { }/[ ] count too). A non-empty array such as [{}] still invokes the target with an empty-payload element — the previous behavior forwarded {} and, for a Lambda target, wrapped it into [{}], i.e. the opposite of AWS. New tests: applyEnrichment_emptyObjectOrArrayResponseSkipsTarget, applyEnrichment_singleElementArrayResponseInvokesTarget.

2. SQS-only wiring — documented explicitly rather than silently bypassing. Enrichment is applied only on the SQS source path; Kinesis/DynamoDB Streams/Kafka deliver straight to the target. Noted at the shared deliverRecords path in code and in a new Enrichment section in docs/services/pipes.md. (Wiring the other pollers can be a follow-up.)

3. Unsupported enrichment types now fail to the DLQ. API destinations, API Gateway and SFN Express are valid enrichment types on AWS but not emulated; applyEnrichment now throws (routed to the pipe DLQ by deliverEnrichedBatch) instead of silently delivering the unenriched payload. New test: applyEnrichment_unsupportedTypeThrows.

The three greptile items were already handled in the current revision: asEventArray is applied only for Lambda targets, and both the enrichment and target Lambda FunctionError paths throw AwsException (routed to the DLQ) rather than a bare RuntimeException / silent success.

Verified (maven, temurin-25): PipesTargetInvokerTest (28) + PipesPollerTest (10) — 38 tests, green.