feat(insight): add FileExporter

insight-plugin/README.md

@@ -28,11 +28,13 @@ 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),
+`FileExporter` (writes to the local filesystem — an EFS mount, an S3 File Gateway path, or `/tmp`).
+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
+`LambdaLogExporter` and `FileExporter` need no extra dependency (`FileExporter` uses only
+`java.nio.file`). The AWS SDK service modules used by the remote exporters are optional so
+applications that use only Lambda logs or the filesystem 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:
 
@@ -52,6 +54,44 @@ application:
 </dependency>
 ```
 
+### FileExporter
+
+Writes records to any writable directory — a Lambda [EFS mount](https://docs.aws.amazon.com/lambda/latest/dg/services-efs.html),
+an S3 File Gateway path, or `/tmp` for local testing. Two modes:
+
+```java
+// NDJSON (default): append one compact JSON line per emission to a date-partitioned file,
+//   {directory}/{YYYY-MM-DD}.ndjson   (date is the record's emittedAt day, UTC)
+.addExporter(FileExporter.builder()
+    .directory("/mnt/efs/workflow-insight")
+    .build())
+
+// JSON: one pretty-printed file per execution, overwritten on each update,
+//   {directory}/{executionName}.json
+.addExporter(FileExporter.builder()
+    .directory("/mnt/efs/workflow-insight")
+    .mode(FileExporter.Mode.JSON)
+    .operationsFormat(FileExporter.OperationsFormat.BOTH) // ARRAY (default) | BY_NAME | BOTH
+    .build())
+```
+
+- **`directory`** is required; the exporter creates it (recursively) on first write.
+- **File names are deterministic and safe.** NDJSON files are named only by UTC date; JSON files use
+  the execution name (falling back to the ARN), with every character outside `[a-zA-Z0-9._-]`
+  replaced by `_`, so a name can never contain a path separator or `..`.
+- **`maxRecordSizeBytes`** has no default — the filesystem has no practical per-record limit. Set it
+  only if you want smaller files; the plugin then truncates each record (oldest results first) to fit
+  before it is written.
+- On `/tmp` (ephemeral, per-container) files do not survive a cold start; use an EFS mount to persist
+  across invocations and containers.
+- **Concurrent NDJSON append is not atomic on shared filesystems.** NDJSON mode uses `Files.write(..., APPEND)`.
+  A single writer on a local disk (`/tmp`) appends one whole line at a time, so lines never interleave. On a
+  **shared NFS/EFS mount written by more than one Lambda environment at once**, the append is not guaranteed
+  atomic: concurrent writers can interleave partial lines or overwrite each other, producing malformed or lost
+  records. If multiple execution environments may write to the same directory, prefer **JSON mode** (one file per
+  execution, keyed by name/ARN — no shared append) or ensure a **single writer** per NDJSON file (for example, a
+  per-environment subdirectory).
+
 ## Design
 
 - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation

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

@@ -3,6 +3,8 @@
 package software.amazon.lambda.durable.insight;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.util.DefaultIndenter;
+import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
@@ -23,6 +25,18 @@ public final class Json {
             .registerModule(new JavaTimeModule())
             .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
 
+    // 2-space indentation with "\n" line breaks, matching JS JSON.stringify(x, null, 2) (LF, no CR). The
+    // DefaultPrettyPrinter is stateful, so build a fresh instance per write via createInstance().
+    private static final DefaultPrettyPrinter PRETTY_PRINTER = buildPrettyPrinter();
+
+    private static DefaultPrettyPrinter buildPrettyPrinter() {
+        DefaultIndenter indenter = new DefaultIndenter("  ", "\n");
+        DefaultPrettyPrinter pp = new DefaultPrettyPrinter();
+        pp.indentObjectsWith(indenter);
+        pp.indentArraysWith(indenter);
+        return pp;
+    }
+
     private Json() {}
 
     public static String stringify(Object value) {
@@ -33,6 +47,15 @@ public static String stringify(Object value) {
         }
     }
 
+    /** Pretty-prints with 2-space indentation (matches the JS {@code JSON.stringify(x, null, 2)} file output). */
+    public static String stringifyPretty(Object value) {
+        try {
+            return MAPPER.writer(PRETTY_PRINTER).writeValueAsString(value);
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("failed to serialize insight record", e);
+        }
+    }
+
     /** UTF-8 byte length of the value's JSON, or {@code null} if it can't be serialized. */
     public static Integer byteSize(Object value) {
         try {

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

@@ -49,6 +49,10 @@ void addOperation(OperationRecord operation) {
         operations.add(operation);
     }
 
+    public String emittedAt() {
+        return emittedAt;
+    }
+
     public String executionArn() {
         return executionArn;
     }
@@ -204,4 +208,22 @@ public Map<String, Object> toByNameWireMap() {
         putTruncationMarkers(data);
         return data;
     }
+
+    /**
+     * Combined wire map mirroring the JS {@code applyOperationsFormat(record, "both")} shape: the canonical
+     * {@code operations} array plus an added {@code operationsByName} map. JS spreads the whole record and then adds
+     * the key ({@code {...record, operationsByName}}), so {@code operationsByName} is the LAST key — after every record
+     * field and any truncation markers. This method preserves that order by starting from the canonical map (which
+     * already ends with the truncation markers) and appending {@code operationsByName} last.
+     */
+    public Map<String, Object> toBothWireMap() {
+        Map<String, Object> data = toWireMap();
+        Map<String, Object> byName = new LinkedHashMap<>();
+        for (Map.Entry<String, OperationSummary> e :
+                OperationsIndex.buildOperationsByName(operations).entrySet()) {
+            byName.put(e.getKey(), e.getValue().toWireMap());
+        }
+        data.put("operationsByName", byName);
+        return data;
+    }
 }

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

@@ -0,0 +1,228 @@
+// 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.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.WorkflowInsightRecord;
+
+/**
+ * Exports workflow insight records to the local filesystem (an EFS mount, an S3 File Gateway path, {@code /tmp} for
+ * testing, or any writable directory). Mirrors the JS {@code FileExporter}.
+ *
+ * <p>Two output modes:
+ *
+ * <ul>
+ *   <li>{@link Mode#NDJSON} (default) — every record is appended as one compact JSON line to a date-partitioned file
+ *       {@code {directory}/{YYYY-MM-DD}.ndjson}, where the date is the record's {@code emittedAt} day (UTC). One line
+ *       per emission, so replay/suspend emissions accumulate.
+ *   <li>{@link Mode#JSON} — each execution gets its own pretty-printed (2-space) file
+ *       {@code {directory}/{executionName}.json}, overwritten on each update so the file always holds the latest view.
+ * </ul>
+ *
+ * <p>File names are derived only from the record's {@code emittedAt} day (NDJSON) or a sanitized execution name/ARN
+ * (JSON): every character outside {@code [a-zA-Z0-9._-]} is replaced with {@code _}, so the name is deterministic and
+ * cannot escape {@code directory} via path separators or {@code ..}. The resolved path is additionally normalized and
+ * verified to stay within {@code directory} as defense in depth.
+ *
+ * <p>The plugin applies {@link #maxRecordSizeBytes()} truncation against {@link #render(WorkflowInsightRecord)} before
+ * {@link #export(WorkflowInsightRecord)} is called; the filesystem has no practical per-record limit, so there is no
+ * default cap. Writes are immediate, so {@link #flush()} is a no-op.
+ *
+ * @see InsightExporter
+ */
+@Experimental
+public final class FileExporter implements InsightExporter {
+
+    /** File output mode. */
+    @Experimental
+    public enum Mode {
+        /** Append every record as one compact JSON line to a date-partitioned {@code .ndjson} file (default). */
+        NDJSON,
+        /** Write one pretty-printed JSON file per execution, overwriting on update. */
+        JSON
+    }
+
+    /** How operations are rendered in the written record. */
+    @Experimental
+    public enum OperationsFormat {
+        /** The canonical {@code operations} array (default). */
+        ARRAY,
+        /** The {@code operationsByName} map, replacing the array. */
+        BY_NAME,
+        /** Both the {@code operations} array and an added {@code operationsByName} map. */
+        BOTH
+    }
+
+    private final Path directory;
+    private final Mode mode;
+    private final OperationsFormat operationsFormat;
+    private final Integer maxRecordSizeBytes;
+    private final AtomicBoolean dirCreated = new AtomicBoolean(false);
+
+    private FileExporter(Builder b) {
+        this.directory = b.directory;
+        this.mode = b.mode != null ? b.mode : Mode.NDJSON;
+        this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY;
+        this.maxRecordSizeBytes = b.maxRecordSizeBytes;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public Integer maxRecordSizeBytes() {
+        return maxRecordSizeBytes;
+    }
+
+    @Override
+    public Object render(WorkflowInsightRecord record) {
+        switch (operationsFormat) {
+            case BY_NAME:
+                return record.toByNameWireMap();
+            case BOTH:
+                return record.toBothWireMap();
+            case ARRAY:
+            default:
+                return record.toWireMap();
+        }
+    }
+
+    @Override
+    public void export(WorkflowInsightRecord record) {
+        ensureDir();
+        Object formatted = render(record);
+        try {
+            if (mode == Mode.NDJSON) {
+                Path file = resolveChild(datePartition(record) + ".ndjson");
+                Files.write(
+                        file,
+                        (Json.stringify(formatted) + "\n").getBytes(StandardCharsets.UTF_8),
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.APPEND);
+            } else {
+                String name = record.executionName() != null ? record.executionName() : record.executionArn();
+                Path file = resolveChild(sanitize(name) + ".json");
+                Files.write(
+                        file,
+                        Json.stringifyPretty(formatted).getBytes(StandardCharsets.UTF_8),
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING);
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException("failed to write insight record to " + directory, e);
+        }
+    }
+
+    /** Writes are immediate; nothing is buffered. */
+    @Override
+    public void flush() {
+        // no-op
+    }
+
+    /** UTC day of the record's {@code emittedAt}, or of "now" when the record carries no {@code emittedAt}. */
+    private static String datePartition(WorkflowInsightRecord record) {
+        String emittedAt = record.emittedAt();
+        if (emittedAt != null && emittedAt.length() >= 10) {
+            return emittedAt.substring(0, 10);
+        }
+        return java.time.Instant.now()
+                .atZone(java.time.ZoneOffset.UTC)
+                .toLocalDate()
+                .toString();
+    }
+
+    private void ensureDir() {
+        if (dirCreated.get()) {
+            return;
+        }
+        try {
+            Files.createDirectories(directory);
+            dirCreated.set(true);
+        } catch (IOException e) {
+            throw new UncheckedIOException("failed to create insight directory " + directory, e);
+        }
+    }
+
+    /**
+     * Resolves a sanitized child file name against {@code directory} and verifies the normalized result stays inside
+     * {@code directory}. The name is already sanitized to {@code [a-zA-Z0-9._-]}, so this is defense in depth against
+     * any future change in the naming rule.
+     */
+    private Path resolveChild(String fileName) {
+        Path base = directory.toAbsolutePath().normalize();
+        Path resolved = base.resolve(fileName).normalize();
+        if (!resolved.startsWith(base)) {
+            throw new IllegalStateException("resolved insight file path escapes the configured directory: " + fileName);
+        }
+        return resolved;
+    }
+
+    /** Replaces every character outside {@code [a-zA-Z0-9._-]} with {@code _} (matches the JS {@code sanitize}). */
+    private static String sanitize(String value) {
+        return value.replaceAll("[^a-zA-Z0-9._-]", "_");
+    }
+
+    /** Builder for {@link FileExporter}. */
+    public static final class Builder {
+        private Path directory;
+        private Mode mode;
+        private OperationsFormat operationsFormat;
+        private Integer maxRecordSizeBytes;
+
+        /** Base directory to write files to, e.g. {@code /mnt/efs/workflow-insight} or {@code /tmp/insight}. */
+        public Builder directory(String directory) {
+            this.directory = directory != null ? Path.of(directory) : null;
+            return this;
+        }
+
+        /** Base directory to write files to. */
+        public Builder directory(Path directory) {
+            this.directory = directory;
+            return this;
+        }
+
+        /** File output mode; defaults to {@link Mode#NDJSON}. */
+        public Builder mode(Mode mode) {
+            this.mode = mode;
+            return this;
+        }
+
+        /** How operations are rendered; defaults to {@link OperationsFormat#ARRAY}. */
+        public Builder operationsFormat(OperationsFormat operationsFormat) {
+            this.operationsFormat = operationsFormat;
+            return this;
+        }
+
+        /**
+         * Max serialized record size before truncation; no default (the filesystem has no practical per-record cap).
+         */
+        public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) {
+            this.maxRecordSizeBytes = maxRecordSizeBytes;
+            return this;
+        }
+
+        public FileExporter build() {
+            if (directory == null) {
+                throw new IllegalArgumentException("directory is required");
+            }
+            if (directory.toString().isBlank()) {
+                throw new IllegalArgumentException("directory must not be blank");
+            }
+            if (maxRecordSizeBytes != null && maxRecordSizeBytes <= 0) {
+                throw new IllegalArgumentException("maxRecordSizeBytes must be positive when set");
+            }
+            return new FileExporter(this);
+        }
+    }
+}