feat(insight): add HttpExporter

insight-plugin/README.md

@@ -28,13 +28,14 @@ DurableConfig config = DurableConfig.builder()
 
 Exporters: `LambdaLogExporter` (default; writes the `operationsByName` map to stdout →
 CloudWatch), `S3Exporter` (canonical `operations` array, one object per execution),
-`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map). Implement
-`InsightExporter` for custom sinks.
+`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map),
+`HttpExporter` (POST/PUT each record to any HTTP(S) endpoint or webhook, choice of operations shape).
+Implement `InsightExporter` for custom sinks.
 
-`LambdaLogExporter` needs no extra dependency. The AWS SDK service modules used by the remote
-exporters are optional so applications that use only Lambda logs do not package them. Add the module
-for each remote exporter you configure, using the AWS SDK for Java 2.x version managed by your
-application:
+`LambdaLogExporter` and `HttpExporter` need no extra dependency — the HTTP exporter uses the JDK's own
+HTTP client. The AWS SDK service modules used by the remote exporters are optional so applications
+that use only Lambda logs or HTTP do not package them. Add the module for each remote exporter you
+configure, using the AWS SDK for Java 2.x version managed by your application:
 
 ```xml
 <!-- Required only for S3Exporter -->
@@ -52,6 +53,43 @@ application:
 </dependency>
 ```
 
+### HttpExporter
+
+`HttpExporter` POSTs (or PUTs) each record to any HTTP(S) endpoint or webhook as a JSON body with a
+default `Content-Type: application/json` header that a caller may override. It uses the JDK's built-in
+HTTP client, so it adds **no dependency**. It mirrors the JS `HttpExporter`.
+
+```java
+.addExporter(HttpExporter.builder()
+    .url("https://collector.example.com/workflow-insight")   // required, absolute http/https URL
+    .method(HttpExporter.Method.POST)                        // POST (default) or PUT
+    .addHeader("Authorization", "Bearer " + token)           // optional auth / API-key headers
+    .timeoutMs(10_000)                                       // per-request timeout, default 10s
+    .operationsFormat(HttpExporter.OperationsFormat.ARRAY)   // ARRAY (default) | BY_NAME | BOTH
+    .build())
+```
+
+Behavior:
+
+- **Endpoint / method / headers.** Records are sent to `url` with `method` (POST or PUT — use PUT for
+  endpoints that upsert by URL path). Your headers are layered on top of the default
+  `Content-Type: application/json`. Header names are matched case-insensitively, so a caller-supplied
+  `Content-Type` (any casing) overrides the default and never produces a duplicate header.
+- **Auth.** There is no built-in auth scheme; pass whatever your endpoint expects as a header
+  (`Authorization: Bearer …`, `x-api-key: …`, etc.) via `addHeader`/`headers`.
+- **Timeout.** `timeoutMs` (default 10000) bounds the whole request; a slow or unreachable endpoint
+  fails fast instead of stalling the invocation. As with every exporter, a failure here is logged and
+  isolated — it never disrupts the durable execution or the other exporters.
+- **Success / failure.** A 2xx response is success; any other status throws (and is contained by the
+  plugin), so a misconfigured endpoint is visible in logs.
+- **Operations shape.** `operationsFormat` selects what the body carries: `ARRAY` (the canonical
+  `operations` array, default), `BY_NAME` (the name-keyed `operationsByName` map), or `BOTH`.
+- **Size cap.** `maxRecordSizeBytes` has no default — a generic HTTP endpoint has no known limit. Set
+  it only if your endpoint caps request size; truncation is then measured against the shape actually
+  sent.
+- **Testing seam.** For unit tests you can inject an `HttpSender` via `.sender(...)` instead of making
+  a real network call, or point `url` at a local in-process server.
+
 ## Design
 
 - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation

insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java

@@ -0,0 +1,240 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.insight.exporters;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import software.amazon.lambda.durable.annotations.Experimental;
+import software.amazon.lambda.durable.insight.InsightExporter;
+import software.amazon.lambda.durable.insight.Json;
+import software.amazon.lambda.durable.insight.OperationSummary;
+import software.amazon.lambda.durable.insight.OperationsIndex;
+import software.amazon.lambda.durable.insight.WorkflowInsightRecord;
+
+/**
+ * Exports workflow insight records to any HTTP(S) endpoint via {@code POST} (or {@code PUT}). Each record is sent as a
+ * JSON body with a default {@code Content-Type: application/json} header that a caller may override; the endpoint must
+ * return a 2xx status or the export throws. Mirrors the JS {@code HttpExporter} contract (URL, method, custom headers,
+ * request timeout, operations format, and optional size cap). Uses the JDK's own HTTP client, so it needs no extra
+ * dependency.
+ */
+@Experimental
+public final class HttpExporter implements InsightExporter {
+
+    /** HTTP method used to deliver the record. */
+    @Experimental
+    public enum Method {
+        POST,
+        PUT
+    }
+
+    /**
+     * How the record's operations are shaped in the posted body: the canonical {@code operations} array, the name-keyed
+     * {@code operationsByName} map, or both. Ports the JS {@code OperationsFormat}.
+     */
+    @Experimental
+    public enum OperationsFormat {
+        ARRAY,
+        BY_NAME,
+        BOTH
+    }
+
+    private final String url;
+    private final Method method;
+    private final Map<String, String> headers;
+    private final Duration timeout;
+    private final OperationsFormat operationsFormat;
+    private final Integer maxRecordSizeBytes;
+    private final HttpSender sender;
+
+    private HttpExporter(Builder b) {
+        this.url = b.url;
+        this.method = b.method != null ? b.method : Method.POST;
+        // Defensive copy so a caller mutating their builder map after build() cannot change this exporter's headers.
+        this.headers = b.headers != null ? new LinkedHashMap<>(b.headers) : new LinkedHashMap<>();
+        this.timeout = b.timeoutMs != null ? Duration.ofMillis(b.timeoutMs) : Duration.ofMillis(10_000);
+        this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY;
+        this.maxRecordSizeBytes = b.maxRecordSizeBytes;
+        this.sender = b.sender != null ? b.sender : new JdkHttpSender();
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public Integer maxRecordSizeBytes() {
+        // No default: a generic HTTP endpoint has no known size limit (matches JS). Truncation is off unless set.
+        return maxRecordSizeBytes;
+    }
+
+    @Override
+    public Object render(WorkflowInsightRecord record) {
+        switch (operationsFormat) {
+            case BY_NAME:
+                return record.toByNameWireMap();
+            case BOTH:
+                return toBothWireMap(record);
+            case ARRAY:
+            default:
+                return record.toWireMap();
+        }
+    }
+
+    /**
+     * The {@code both} shape: the canonical array wire map with an added {@code operationsByName} map. Composed here
+     * from the record's public renderings so the SDK core needs no new method. The {@code operationsByName} entry is
+     * placed before the trailing truncation markers to keep a stable, readable field order.
+     */
+    private Map<String, Object> toBothWireMap(WorkflowInsightRecord record) {
+        Map<String, Object> data = new LinkedHashMap<>(record.toWireMap());
+        Map<String, Object> byName = new LinkedHashMap<>();
+        for (Map.Entry<String, OperationSummary> e :
+                OperationsIndex.buildOperationsByName(record.operations()).entrySet()) {
+            byName.put(e.getKey(), e.getValue().toWireMap());
+        }
+        Map<String, Object> merged = new LinkedHashMap<>();
+        for (Map.Entry<String, Object> e : data.entrySet()) {
+            // Insert operationsByName immediately after the operations array, before truncation markers.
+            if ("truncated".equals(e.getKey()) && !merged.containsKey("operationsByName")) {
+                merged.put("operationsByName", byName);
+            }
+            merged.put(e.getKey(), e.getValue());
+        }
+        merged.putIfAbsent("operationsByName", byName);
+        return merged;
+    }
+
+    @Override
+    public void export(WorkflowInsightRecord record) {
+        String body = Json.stringify(render(record));
+        HttpSender.Response response = sender.send(url, method.name(), mergeHeaders(), body, timeout);
+        int status = response.statusCode();
+        if (status < 200 || status >= 300) {
+            throw new IllegalStateException(
+                    "HttpExporter: endpoint returned " + status + " " + response.reasonPhrase());
+        }
+    }
+
+    /**
+     * Builds the request headers: the default {@code Content-Type: application/json} with the caller's headers layered
+     * on top. Header names are matched case-insensitively (HTTP header names are case-insensitive), so a caller header
+     * of any casing — {@code content-type}, {@code CONTENT-TYPE}, {@code Content-Type} — replaces an earlier entry
+     * instead of adding a duplicate. The result carries exactly one header per name, and any caller-supplied
+     * {@code Content-Type} overrides the default.
+     */
+    private Map<String, String> mergeHeaders() {
+        Map<String, String> merged = new LinkedHashMap<>();
+        merged.put("Content-Type", "application/json");
+        for (Map.Entry<String, String> e : headers.entrySet()) {
+            merged.keySet().removeIf(existing -> existing.equalsIgnoreCase(e.getKey()));
+            merged.put(e.getKey(), e.getValue());
+        }
+        return merged;
+    }
+
+    /** Builder for {@link HttpExporter}. */
+    public static final class Builder {
+        private String url;
+        private Method method;
+        private Map<String, String> headers;
+        private Integer timeoutMs;
+        private OperationsFormat operationsFormat;
+        private Integer maxRecordSizeBytes;
+        private HttpSender sender;
+
+        /** Endpoint to POST/PUT records to (required). Must be an absolute {@code http} or {@code https} URL. */
+        public Builder url(String url) {
+            this.url = url;
+            return this;
+        }
+
+        /** HTTP method. Default {@link Method#POST}. Use {@link Method#PUT} for endpoints that upsert by URL path. */
+        public Builder method(Method method) {
+            this.method = method;
+            return this;
+        }
+
+        /**
+         * Additional request headers (for example an {@code Authorization} token or API key). Copied defensively.
+         * Header names are matched case-insensitively, and a header named {@code Content-Type} (any casing) overrides
+         * the exporter's default {@code application/json}.
+         */
+        public Builder headers(Map<String, String> headers) {
+            this.headers = headers;
+            return this;
+        }
+
+        /**
+         * Adds a single request header; convenient for one auth header without building a map. Follows the same
+         * case-insensitive, default-overriding rules as {@link #headers(Map)}.
+         */
+        public Builder addHeader(String name, String value) {
+            if (this.headers == null) {
+                this.headers = new LinkedHashMap<>();
+            }
+            this.headers.put(name, value);
+            return this;
+        }
+
+        /** Request timeout in milliseconds. Default 10000 (10s). Must be positive. */
+        public Builder timeoutMs(Integer timeoutMs) {
+            this.timeoutMs = timeoutMs;
+            return this;
+        }
+
+        /** How operations are rendered in the posted body. Default {@link OperationsFormat#ARRAY}. */
+        public Builder operationsFormat(OperationsFormat operationsFormat) {
+            this.operationsFormat = operationsFormat;
+            return this;
+        }
+
+        /** Max serialized record size before truncation. No default; set it if your endpoint caps request size. */
+        public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) {
+            this.maxRecordSizeBytes = maxRecordSizeBytes;
+            return this;
+        }
+
+        /** Test seam: inject an {@link HttpSender} instead of the default JDK HTTP client. */
+        public Builder sender(HttpSender sender) {
+            this.sender = sender;
+            return this;
+        }
+
+        public HttpExporter build() {
+            validate();
+            return new HttpExporter(this);
+        }
+
+        private void validate() {
+            if (url == null || url.isBlank()) {
+                throw new IllegalArgumentException("HttpExporter requires a url");
+            }
+            URI uri;
+            try {
+                uri = new URI(url);
+            } catch (URISyntaxException e) {
+                throw new IllegalArgumentException("HttpExporter url is not a valid URI: " + url, e);
+            }
+            String scheme = uri.getScheme();
+            if (scheme == null || uri.getHost() == null) {
+                throw new IllegalArgumentException("HttpExporter url must be absolute with a host: " + url);
+            }
+            String lower = scheme.toLowerCase(Locale.ROOT);
+            if (!"http".equals(lower) && !"https".equals(lower)) {
+                throw new IllegalArgumentException("HttpExporter url scheme must be http or https: " + url);
+            }
+            if (timeoutMs != null && timeoutMs <= 0) {
+                throw new IllegalArgumentException("HttpExporter timeoutMs must be positive: " + timeoutMs);
+            }
+            if (maxRecordSizeBytes != null && maxRecordSizeBytes <= 0) {
+                throw new IllegalArgumentException(
+                        "HttpExporter maxRecordSizeBytes must be positive: " + maxRecordSizeBytes);
+            }
+        }
+    }
+}

insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpSender.java

@@ -0,0 +1,44 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.insight.exporters;
+
+import java.time.Duration;
+import java.util.Map;
+import software.amazon.lambda.durable.annotations.Experimental;
+
+/**
+ * Test/injection seam for {@link HttpExporter}'s HTTP transport. The default implementation ({@link JdkHttpSender}) is
+ * backed by {@link java.net.http.HttpClient}; tests inject a stub or point the exporter at a local in-process server.
+ * Mirrors the {@code client(...)} builder seam on the AWS SDK exporters, but avoids adding any dependency by staying on
+ * the JDK's own HTTP client.
+ */
+@Experimental
+public interface HttpSender {
+
+    /**
+     * Sends one request and returns the response status. Implementations MUST apply {@code timeout} to the request and
+     * MUST throw on transport failure (I/O error, timeout, interruption) so the exporter surfaces it like the JS
+     * reference (whose {@code fetch} rejects). {@code headers} already includes {@code Content-Type: application/json}.
+     */
+    Response send(String url, String method, Map<String, String> headers, String body, Duration timeout);
+
+    /** The parts of an HTTP response the exporter needs to decide success/failure. */
+    @Experimental
+    final class Response {
+        private final int statusCode;
+        private final String reasonPhrase;
+
+        public Response(int statusCode, String reasonPhrase) {
+            this.statusCode = statusCode;
+            this.reasonPhrase = reasonPhrase != null ? reasonPhrase : "";
+        }
+
+        public int statusCode() {
+            return statusCode;
+        }
+
+        public String reasonPhrase() {
+            return reasonPhrase;
+        }
+    }
+}