etdl-target-java 0.0.2

//! The `java` [`CodeGenerator`] target.
//!
//! Consumes the exact same validated [`EtlDocument`] + resolved fault-tree
//! probabilities + [`AsyncApiRegistry`] that [`etdl_compiler::RustCodeGenerator`]
//! consumes — nothing in this crate re-parses `.etdl`, re-validates ECEL
//! conditions, or re-evaluates fault trees. Only the code-generation layer
//! differs. See `docs/architecture/targets.md` for the full picture.
//!
//! Building this crate never requires a JDK — it only emits Java source
//! text. A JDK is needed solely to *compile the generated output*, which is
//! entirely the concern of whoever consumes the generated files.
//!
//! # Output shape
//!
//! - `etdl/runtime/*.java` — a small, fixed, read-only Java runtime
//!   (`WorkflowError`, `Publisher`, `BranchMonitor`, `RetryPolicy`,
//!   `BackoffStrategy`, `Condition`) mirroring the pieces of `etdl-core`
//!   every generated Rust file already depends on. Not document-specific;
//!   identical across every `--target java` invocation.
//! - `<package>/<Type>.java` — one record file per distinct AsyncAPI message
//!   referenced anywhere in the document (both External and Internal
//!   References resolve through [`AsyncApiRegistry::resolve_message`], so —
//!   unlike the Rust target — Java never assumes a pre-existing
//!   `{alias}::messages`-equivalent package exists for an External
//!   Reference; it generates the record either way).
//! - `<package>/<Stem>Workflow.java` — one `public final class` holding the
//!   fault-tree-probability constants and one `public static` orchestration
//!   method per event tree — the stable, read-only, autogenerated
//!   orchestration logic (branches, retries, SLA/branch monitoring, sends),
//!   structurally the same thing `RustCodeGenerator::generate_event_tree_handler`
//!   emits, translated to Java control flow.
//! - `<package>/<Tree>Handlers.java` — one `public interface` per event
//!   tree, one method per distinct `handler:` id used in that tree. This is
//!   the user-editable/application-specific boundary: the orchestration
//!   class calls through this interface, never the other way around, so a
//!   developer implements it in their own (never regenerated) class without
//!   ever touching the generated orchestration or runtime code — the same
//!   role Rust's naming-convention `async fn <handler>(...)` functions play,
//!   made explicit as a generated contract because Java has no equivalent
//!   free-function-by-convention mechanism.
//!
//! # Known scope limits (see the crate-level deliverables report)
//!
//! Generated Java carries no JSON library dependency (unlike the Rust
//! target, which bakes in `serde`/`serde_json`): message records are plain
//! data carriers and `Publisher.publish` takes `Object`. (De)serialization
//! across the wire is left to the integration layer the developer wires up,
//! so the generated output — and this crate's own compile-check test —
//! never require a JSON library to be on the classpath, only a JDK.

use etdl_compiler::{CodeGenerator, Diagnostic, GeneratedFile};
use etdl_parser::ast::{
    BackoffStrategy as AstBackoffStrategy, ChannelRef, Condition, EtlDocument, EventTree,
    MessageRef, Node,
};
use etdl_parser::asyncapi::AsyncApiRegistry;
use etdl_parser::ecel;
use std::collections::BTreeMap;

pub struct JavaCodeGenerator {
    pub version: String,
    /// Java package for document-specific output (`<package>/<Type>.java`,
    /// `<package>/<Stem>Workflow.java`, `<package>/<Tree>Handlers.java`).
    /// `None` derives it from `info.domain` (falling back to
    /// `etdl.generated` when that doesn't yield a usable package name).
    /// The fixed `etdl.runtime` package is never affected by this field.
    pub package: Option<String>,
}

impl Default for JavaCodeGenerator {
    fn default() -> Self {
        Self::new()
    }
}

impl JavaCodeGenerator {
    pub fn new() -> Self {
        JavaCodeGenerator {
            version: env!("CARGO_PKG_VERSION").to_string(),
            package: None,
        }
    }

    fn resolve_package(&self, doc: &EtlDocument) -> String {
        if let Some(ref p) = self.package {
            return p.clone();
        }
        sanitize_package(&doc.info.domain)
    }
}

impl CodeGenerator for JavaCodeGenerator {
    fn target_name(&self) -> &'static str {
        "java"
    }

    fn generate_all(
        &self,
        doc: &EtlDocument,
        fault_tree_probs: &BTreeMap<String, f64>,
        registry: &AsyncApiRegistry,
        stem: &str,
        _diagnostics: &mut Vec<Diagnostic>,
    ) -> Result<Vec<GeneratedFile>, String> {
        let package = self.resolve_package(doc);
        let package_dir = package.replace('.', "/");
        let class_stem = to_pascal_case(&sanitize_ident(stem));

        let mut files = runtime_files();

        let message_refs = collect_message_refs(doc);
        for (type_name, msg_ref) in &message_refs {
            let value = registry
                .resolve_message(doc, msg_ref)
                .map_err(|e| format!("resolving message for '{}': {}", type_name, e))?;
            let contents = generate_message_file(&package, type_name, &value);
            files.push(GeneratedFile::new(
                format!("{package_dir}/{type_name}.java"),
                contents,
            ));
        }

        for tree in doc.event_trees.values() {
            let tree_class = to_pascal_case(&sanitize_ident(tree_id_of(doc, tree)));
            let handlers_iface = format!("{tree_class}Handlers");
            let handler_methods = collect_handler_methods(tree);
            let msg_type = ref_to_type_name(&tree.initiating_event.message);

            files.push(GeneratedFile::new(
                format!("{package_dir}/{handlers_iface}.java"),
                generate_handlers_interface(&package, &handlers_iface, &msg_type, &handler_methods),
            ));
        }

        let workflow = generate_workflow_class(doc, &package, &class_stem, fault_tree_probs)?;
        files.push(GeneratedFile::new(
            format!("{package_dir}/{class_stem}Workflow.java"),
            workflow,
        ));

        Ok(files)
    }
}

fn tree_id_of<'a>(doc: &'a EtlDocument, tree: &EventTree) -> &'a str {
    doc.event_trees
        .iter()
        .find(|(_, t)| std::ptr::eq(*t, tree))
        .map(|(id, _)| id.as_str())
        .unwrap_or("Tree")
}

// ---------------------------------------------------------------------
// Fixed, document-independent runtime (`etdl/runtime/*.java`)
// ---------------------------------------------------------------------

fn runtime_files() -> Vec<GeneratedFile> {
    vec![
        GeneratedFile::new("etdl/runtime/WorkflowError.java", WORKFLOW_ERROR_JAVA),
        GeneratedFile::new("etdl/runtime/Publisher.java", PUBLISHER_JAVA),
        GeneratedFile::new("etdl/runtime/BackoffStrategy.java", BACKOFF_STRATEGY_JAVA),
        GeneratedFile::new("etdl/runtime/EtdlNative.java", ETDL_NATIVE_JAVA),
        GeneratedFile::new("etdl/runtime/BranchMonitor.java", BRANCH_MONITOR_JAVA),
        GeneratedFile::new("etdl/runtime/RetryPolicy.java", RETRY_POLICY_JAVA),
        GeneratedFile::new("etdl/runtime/Condition.java", CONDITION_JAVA),
    ]
}

const WORKFLOW_ERROR_JAVA: &str = r#"package etdl.runtime;

/** Read-only ETDL runtime. DO NOT EDIT DIRECTLY. */
public class WorkflowError extends Exception {
    public WorkflowError(String message) {
        super(message);
    }

    public WorkflowError(String message, Throwable cause) {
        super(message, cause);
    }
}
"#;

const PUBLISHER_JAVA: &str = r#"package etdl.runtime;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * The stable API boundary generated orchestration code calls into for a
 * `consequence: send` (core spec Section 5.5.3) — the Java analog of
 * `etdl_core::publisher::Publisher`. A developer supplies a concrete
 * implementation (wiring it to a real broker/client and handling
 * serialization however fits their stack); generated code never depends on
 * a specific transport or JSON library. This is application integration,
 * not ETDL runtime machinery, so unlike `BranchMonitor`/`RetryPolicy`/
 * `Condition` it has no native binding of its own — there is no shared
 * "publish" semantics in the Rust runtime to delegate to.
 */
public interface Publisher {
    void publish(String channel, Object payload) throws WorkflowError;
}
"#;

const BACKOFF_STRATEGY_JAVA: &str = r#"package etdl.runtime;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * Mirrors `etdl_runtime_ffi`'s `strategy` parameter encoding exactly
 * (`nativeValue()`): `0` = fixed, `1` = exponential. Keep in sync with
 * `etdl-runtime-ffi/src/lib.rs`'s `etdl_retry_policy_new`.
 */
public enum BackoffStrategy {
    FIXED(0),
    EXPONENTIAL(1);

    private final int nativeValue;

    BackoffStrategy(int nativeValue) {
        this.nativeValue = nativeValue;
    }

    public int nativeValue() {
        return nativeValue;
    }
}
"#;

/// The native-binding layer: loads `libetdl_runtime_ffi` and exposes every
/// function from `etdl-runtime-ffi/src/lib.rs` as a `MethodHandle`, using
/// Java 21's (preview) Foreign Function & Memory API — no JNI glue code,
/// no per-language native crate, just a direct binding to the same plain C
/// ABI Go/.NET/Python bind to as well. `BranchMonitor`/`RetryPolicy`/
/// `Condition` are thin facades over this; nothing here re-implements ETDL
/// semantics — every method body is a native call plus argument/return
/// marshaling.
const ETDL_NATIVE_JAVA: &str = r#"package etdl.runtime;

import java.lang.foreign.Arena;
import java.lang.foreign.FunctionDescriptor;
import java.lang.foreign.Linker;
import java.lang.foreign.MemorySegment;
import java.lang.foreign.SymbolLookup;
import java.lang.invoke.MethodHandle;

import static java.lang.foreign.ValueLayout.ADDRESS;
import static java.lang.foreign.ValueLayout.JAVA_BOOLEAN;
import static java.lang.foreign.ValueLayout.JAVA_DOUBLE;
import static java.lang.foreign.ValueLayout.JAVA_INT;
import static java.lang.foreign.ValueLayout.JAVA_LONG;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * Binds to `libetdl_runtime_ffi` (built from the `etdl-runtime-ffi` Rust
 * crate — see `docs/architecture/targets.md`). Requires the JVM to run
 * with `--enable-preview --enable-native-access=ALL-UNNAMED` (Java 21's
 * Foreign Function & Memory API is a preview feature; it is finalized,
 * flag-free, in Java 22+).
 *
 * Library resolution order:
 * 1. system property `etdl.runtime.library` — an explicit absolute path
 *    to the shared library (`libetdl_runtime_ffi.so`/`.dylib`,
 *    `etdl_runtime_ffi.dll`).
 * 2. `System.loadLibrary("etdl_runtime_ffi")` — standard JVM native
 *    library search (`-Djava.library.path=...`, or the platform's usual
 *    shared-library search path).
 */
final class EtdlNative {
    private EtdlNative() {}

    // Mirrors etdl-runtime-ffi's ETDL_OK / ETDL_ERR_* / ETDL_RETRY_* constants exactly.
    static final int OK = 0;
    static final int ERR_NULL_HANDLE = -1;
    static final int ERR_INVALID_ARG = -2;
    static final int RETRY_OK = 0;
    static final int RETRY_EXHAUSTED = 1;
    static final int RETRY_FATAL = 2;

    private static final Linker LINKER = Linker.nativeLinker();
    private static final SymbolLookup LOOKUP = load();

    static final MethodHandle ABI_VERSION =
            handle("etdl_runtime_abi_version", FunctionDescriptor.of(JAVA_INT));
    static final MethodHandle LAST_ERROR_MESSAGE =
            handle("etdl_last_error_message", FunctionDescriptor.of(ADDRESS));
    static final MethodHandle STRING_FREE =
            handle("etdl_string_free", FunctionDescriptor.ofVoid(ADDRESS));
    static final MethodHandle BRANCH_MONITOR_NEW =
            handle("etdl_branch_monitor_new", FunctionDescriptor.of(ADDRESS, ADDRESS));
    static final MethodHandle BRANCH_MONITOR_FREE =
            handle("etdl_branch_monitor_free", FunctionDescriptor.ofVoid(ADDRESS));
    static final MethodHandle BRANCH_MONITOR_RECORD_BRANCH = handle(
            "etdl_branch_monitor_record_branch",
            FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_DOUBLE));
    static final MethodHandle BRANCH_MONITOR_RECORD_SUCCESS = handle(
            "etdl_branch_monitor_record_success",
            FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_DOUBLE, JAVA_BOOLEAN));
    static final MethodHandle BRANCH_MONITOR_RECORD_FAILURE = handle(
            "etdl_branch_monitor_record_failure",
            FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS, JAVA_DOUBLE, JAVA_BOOLEAN));
    static final MethodHandle RETRY_POLICY_NEW = handle(
            "etdl_retry_policy_new",
            FunctionDescriptor.of(ADDRESS, JAVA_INT, JAVA_LONG, JAVA_INT));
    static final MethodHandle RETRY_POLICY_FREE =
            handle("etdl_retry_policy_free", FunctionDescriptor.ofVoid(ADDRESS));
    static final MethodHandle RETRY_POLICY_EXECUTE = handle(
            "etdl_retry_policy_execute",
            FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS, ADDRESS));
    static final MethodHandle CONDITION_MATCHES =
            handle("etdl_condition_matches", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS));
    static final MethodHandle CONDITION_CONTAINS =
            handle("etdl_condition_contains", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS));

    private static SymbolLookup load() {
        String explicitPath = System.getProperty("etdl.runtime.library");
        if (explicitPath != null) {
            return SymbolLookup.libraryLookup(explicitPath, Arena.global());
        }
        String mappedName = System.mapLibraryName("etdl_runtime_ffi");
        try {
            System.loadLibrary("etdl_runtime_ffi");
        } catch (UnsatisfiedLinkError e) {
            throw new UnsatisfiedLinkError(
                    "could not load the ETDL Rust runtime ('" + mappedName + "'). Build it with "
                            + "`cargo build -p etdl-runtime-ffi --release` and either put its "
                            + "output directory on -Djava.library.path=..., or point "
                            + "-Detdl.runtime.library=<absolute path to " + mappedName
                            + "> directly at the file. Original error: " + e.getMessage());
        }
        return SymbolLookup.loaderLookup();
    }

    private static MethodHandle handle(String name, FunctionDescriptor descriptor) {
        MemorySegment symbol = LOOKUP.find(name).orElseThrow(() -> new UnsatisfiedLinkError(
                "symbol not found in the ETDL Rust runtime: " + name
                        + " (etdl-runtime-ffi version mismatch?)"));
        return LINKER.downcallHandle(symbol, descriptor);
    }

    /** The native ABI version (`etdl_runtime_abi_version()`), for diagnostics. */
    static int abiVersion() {
        return invokeInt(ABI_VERSION);
    }

    /**
     * Reads and frees the calling thread's last native error message (see
     * `etdl_last_error_message`/`etdl_string_free`), or a fallback string
     * if none is set.
     */
    static String lastError() {
        try {
            MemorySegment ptr = (MemorySegment) LAST_ERROR_MESSAGE.invokeWithArguments();
            if (MemorySegment.NULL.equals(ptr)) {
                return "(no further detail available from the native runtime)";
            }
            String message = ptr.reinterpret(4096).getUtf8String(0);
            STRING_FREE.invokeWithArguments(ptr);
            return message;
        } catch (Throwable t) {
            return "(failed to read native error message: " + t + ")";
        }
    }

    static MemorySegment cString(Arena arena, String s) {
        return arena.allocateUtf8String(s);
    }

    static int invokeInt(MethodHandle mh, Object... args) {
        try {
            return (int) mh.invokeWithArguments(args);
        } catch (Throwable t) {
            throw new RuntimeException("native ETDL runtime call failed: " + t.getMessage(), t);
        }
    }

    static MemorySegment invokeAddress(MethodHandle mh, Object... args) {
        try {
            return (MemorySegment) mh.invokeWithArguments(args);
        } catch (Throwable t) {
            throw new RuntimeException("native ETDL runtime call failed: " + t.getMessage(), t);
        }
    }

    static void invokeVoid(MethodHandle mh, Object... args) {
        try {
            mh.invokeWithArguments(args);
        } catch (Throwable t) {
            throw new RuntimeException("native ETDL runtime call failed: " + t.getMessage(), t);
        }
    }

    /**
     * Builds a native-callable upcall stub for `target` — the mechanism
     * {@link RetryPolicy#execute} uses to let `etdl_retry_policy_execute`'s
     * native loop call back into Java per attempt. `arena`-scoped: the stub
     * is only valid for `arena`'s lifetime, matching a single `execute`
     * call.
     */
    static MemorySegment upcallStub(MethodHandle target, FunctionDescriptor descriptor, Arena arena) {
        return LINKER.upcallStub(target, descriptor, arena);
    }
}
"#;

/// The Java analog of `etdl_core::BranchMonitor` — but not a
/// reimplementation of it: every method here is a direct call into the
/// same `BranchMonitor` the Rust target's generated code has always used,
/// via `etdl-runtime-ffi`. SLA anomaly detection, chaos injection, and
/// observation emission all happen exactly once, in Rust, for every
/// language target alike.
const BRANCH_MONITOR_JAVA: &str = r#"package etdl.runtime;

import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * Thin facade over the native `etdl_core::BranchMonitor` (via
 * `etdl-runtime-ffi`). Owns one native handle; always use in
 * try-with-resources (generated orchestration code always does) so
 * `etdl_branch_monitor_free` runs deterministically.
 */
public final class BranchMonitor implements AutoCloseable {
    private final Arena arena = Arena.ofConfined();
    private final MemorySegment handle;
    private boolean closed = false;

    public BranchMonitor(String nodeId) {
        MemorySegment nodeIdSeg = EtdlNative.cString(arena, nodeId);
        this.handle = EtdlNative.invokeAddress(EtdlNative.BRANCH_MONITOR_NEW, nodeIdSeg);
        if (MemorySegment.NULL.equals(this.handle)) {
            arena.close();
            throw new IllegalStateException(
                    "etdl_branch_monitor_new failed: " + EtdlNative.lastError());
        }
    }

    public void recordBranch(String outcome, double probability) {
        try (Arena call = Arena.ofConfined()) {
            MemorySegment outcomeSeg = EtdlNative.cString(call, outcome);
            int rc = EtdlNative.invokeInt(
                    EtdlNative.BRANCH_MONITOR_RECORD_BRANCH, handle, outcomeSeg, probability);
            checkRc(rc, "etdl_branch_monitor_record_branch");
        }
    }

    /** `probability` may be `null` when the operation has no linked fault tree. */
    public void recordSuccess(String operationId, Double probability) {
        try (Arena call = Arena.ofConfined()) {
            MemorySegment opSeg = EtdlNative.cString(call, operationId);
            int rc = EtdlNative.invokeInt(
                    EtdlNative.BRANCH_MONITOR_RECORD_SUCCESS,
                    handle, opSeg, probability == null ? 0.0 : probability, probability != null);
            checkRc(rc, "etdl_branch_monitor_record_success");
        }
    }

    /** `probability` may be `null` when the operation has no linked fault tree. */
    public void recordFailure(String operationId, Throwable error, Double probability) {
        try (Arena call = Arena.ofConfined()) {
            MemorySegment opSeg = EtdlNative.cString(call, operationId);
            MemorySegment errSeg = EtdlNative.cString(call, String.valueOf(error.getMessage()));
            int rc = EtdlNative.invokeInt(
                    EtdlNative.BRANCH_MONITOR_RECORD_FAILURE,
                    handle, opSeg, errSeg, probability == null ? 0.0 : probability, probability != null);
            checkRc(rc, "etdl_branch_monitor_record_failure");
        }
    }

    private void checkRc(int rc, String call) {
        if (rc != EtdlNative.OK) {
            throw new RuntimeException(call + " failed (code " + rc + "): " + EtdlNative.lastError());
        }
    }

    @Override
    public void close() {
        if (!closed) {
            closed = true;
            EtdlNative.invokeVoid(EtdlNative.BRANCH_MONITOR_FREE, handle);
            arena.close();
        }
    }
}
"#;

/// The Java analog of `etdl_core::retry::RetryPolicy` — the attempt-count
/// and backoff *sequence* are computed once, in Rust
/// (`etdl_retry_policy_execute`'s native loop), via a callback bound to
/// `op`. Per-attempt timeout enforcement stays here (see
/// `docs/architecture/targets.md`'s "Ownership model": no FFI boundary can
/// safely cancel foreign code mid-flight), using a virtual-thread executor
/// exactly like this class did before it called into Rust at all.
const RETRY_POLICY_JAVA: &str = r#"package etdl.runtime;

import java.lang.foreign.Arena;
import java.lang.foreign.FunctionDescriptor;
import java.lang.foreign.MemorySegment;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static java.lang.foreign.ValueLayout.ADDRESS;
import static java.lang.foreign.ValueLayout.JAVA_INT;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * Thin facade over the native `etdl_core::RetryPolicy` (via
 * `etdl-runtime-ffi`'s callback-driven `etdl_retry_policy_execute`). Owns
 * one native handle; always use in try-with-resources.
 */
public final class RetryPolicy implements AutoCloseable {
    @FunctionalInterface
    public interface Op<T> {
        T call() throws WorkflowError;
    }

    private final Arena arena = Arena.ofConfined();
    private final MemorySegment handle;
    private boolean closed = false;

    public RetryPolicy(int maxAttempts, long backoffMs, BackoffStrategy strategy) {
        MemorySegment h = EtdlNative.invokeAddress(
                EtdlNative.RETRY_POLICY_NEW, maxAttempts, backoffMs, strategy.nativeValue());
        if (MemorySegment.NULL.equals(h)) {
            arena.close();
            throw new IllegalStateException("etdl_retry_policy_new failed: " + EtdlNative.lastError());
        }
        this.handle = h;
    }

    /**
     * Trampoline bound (per `execute` call, via {@link MethodHandles.Lookup#bind})
     * to a native upcall stub — this is what `etdl_retry_policy_execute`'s
     * `attempt` loop actually calls back into on every attempt. Runs `op`
     * under `timeoutMs`; its return value is the contract
     * `etdl_retry_policy_execute` documents: `0` success, positive
     * retryable, negative fatal. `op`'s own exception (checked
     * `WorkflowError` or any unchecked exception) is caught here — it never
     * escapes back across the native call, exactly as the FFI boundary
     * requires.
     */
    private final class Trampoline {
        private final Op<?> op;
        private final long timeoutMs;
        Object result;
        WorkflowError checkedError;
        RuntimeException uncheckedError;

        Trampoline(Op<?> op, long timeoutMs) {
            this.op = op;
            this.timeoutMs = timeoutMs;
        }

        @SuppressWarnings("unused") // invoked only via the bound MethodHandle below
        int invoke(MemorySegment userData, int attempt) {
            try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
                Future<?> future = executor.submit(op::call);
                try {
                    result = future.get(timeoutMs, TimeUnit.MILLISECONDS);
                    return 0;
                } catch (TimeoutException te) {
                    future.cancel(true);
                    checkedError = new WorkflowError("operation timed out after " + timeoutMs + "ms", te);
                    return 1;
                } catch (ExecutionException ee) {
                    Throwable cause = ee.getCause();
                    if (cause instanceof WorkflowError we) {
                        checkedError = we;
                        return 1;
                    }
                    uncheckedError = (cause instanceof RuntimeException re) ? re : new RuntimeException(cause);
                    return -1;
                }
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                uncheckedError = new RuntimeException(ie);
                return -1;
            } catch (RuntimeException re) {
                // Defensive: a panic-equivalent from `op` itself escaping the
                // executor submission path, not just its Future. Never let it
                // propagate through the native call.
                uncheckedError = re;
                return -1;
            }
        }
    }

    public <T> T execute(Op<T> op, long timeoutMs) throws WorkflowError {
        Trampoline trampoline = new Trampoline(op, timeoutMs);
        MethodHandle bound;
        try {
            bound = MethodHandles.lookup().bind(
                    trampoline, "invoke", MethodType.methodType(int.class, MemorySegment.class, int.class));
        } catch (ReflectiveOperationException e) {
            throw new AssertionError("Trampoline.invoke signature must match the callback descriptor", e);
        }

        try (Arena call = Arena.ofConfined()) {
            MemorySegment callbackStub = EtdlNative.upcallStub(
                    bound, FunctionDescriptor.of(JAVA_INT, ADDRESS, JAVA_INT), call);
            MemorySegment outAttempts = call.allocate(JAVA_INT);
            int rc = EtdlNative.invokeInt(
                    EtdlNative.RETRY_POLICY_EXECUTE, handle, callbackStub, MemorySegment.NULL, outAttempts);

            if (rc == EtdlNative.RETRY_OK) {
                @SuppressWarnings("unchecked")
                T result = (T) trampoline.result;
                return result;
            }
            if (trampoline.uncheckedError != null) {
                throw trampoline.uncheckedError;
            }
            if (trampoline.checkedError != null) {
                throw trampoline.checkedError;
            }
            throw new WorkflowError(
                    (rc == EtdlNative.RETRY_FATAL ? "retry failed fatally" : "retry exhausted")
                            + ": " + EtdlNative.lastError());
        }
    }

    @Override
    public void close() {
        if (!closed) {
            closed = true;
            EtdlNative.invokeVoid(EtdlNative.RETRY_POLICY_FREE, handle);
            arena.close();
        }
    }
}
"#;

/// The Java analog of `etdl_core::condition::{contains, matches}` — ECEL
/// `in`/`matches` semantics (RE2-compatible regex, set membership) stay in
/// Rust; this facade only marshals `List`/scalar values to the JSON text
/// `etdl_condition_contains` expects (a small, dependency-free encoder —
/// see its doc comment for why generated Java carries no JSON library).
const CONDITION_JAVA: &str = r#"package etdl.runtime;

import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.util.List;

/**
 * Read-only ETDL runtime. DO NOT EDIT DIRECTLY.
 *
 * ECEL (core spec Section 6) `in`/`matches` operators lower to these
 * helpers, delegating to the native `etdl_core::condition` module (via
 * `etdl-runtime-ffi`) — so a `matches` regex or an `in` membership check
 * behaves byte-for-byte identically no matter which target (Rust, Java,
 * Python, Go, .NET) evaluates it.
 */
public final class Condition {
    private Condition() {}

    public static boolean matches(String value, String pattern) {
        try (Arena arena = Arena.ofConfined()) {
            MemorySegment valueSeg = EtdlNative.cString(arena, value == null ? "" : value);
            MemorySegment patternSeg = EtdlNative.cString(arena, pattern);
            int rc = EtdlNative.invokeInt(EtdlNative.CONDITION_MATCHES, valueSeg, patternSeg);
            if (rc < 0) {
                throw new IllegalArgumentException("etdl_condition_matches: " + EtdlNative.lastError());
            }
            return rc == 1;
        }
    }

    /**
     * `length()` (core spec Section 6.5.1): the codepoint count of a
     * string, or the element count of an array. Takes {@code Object}
     * deliberately — a payload field's generated accessor has a concrete
     * static type (e.g. {@code String}, never {@code Object}), and {@code
     * instanceof List<?>} against a provably-unrelated static type (e.g.
     * {@code String}) is a compile error in Java's flow-sensitive pattern
     * matching, not just a false branch. Widening the call site's argument
     * to {@code Object} first sidesteps that entirely.
     */
    public static double length(Object value) {
        if (value instanceof List<?> list) {
            return list.size();
        }
        return value == null ? 0 : value.toString().length();
    }

    public static boolean contains(List<?> haystack, Object needle) {
        try (Arena arena = Arena.ofConfined()) {
            MemorySegment needleSeg = EtdlNative.cString(arena, toJsonScalar(needle));
            MemorySegment haystackSeg = EtdlNative.cString(arena, toJsonArray(haystack));
            int rc = EtdlNative.invokeInt(EtdlNative.CONDITION_CONTAINS, needleSeg, haystackSeg);
            if (rc < 0) {
                throw new IllegalArgumentException("etdl_condition_contains: " + EtdlNative.lastError());
            }
            return rc == 1;
        }
    }

    /**
     * A minimal, dependency-free JSON array encoder — deliberately not a
     * general-purpose JSON library, only what ECEL `in`'s scalar-typed
     * elements (string/number/boolean/null) need to reach
     * `etdl_condition_contains`'s JSON-text wire format.
     */
    private static String toJsonArray(List<?> items) {
        StringBuilder sb = new StringBuilder("[");
        if (items != null) {
            for (int i = 0; i < items.size(); i++) {
                if (i > 0) {
                    sb.append(',');
                }
                sb.append(toJsonScalar(items.get(i)));
            }
        }
        return sb.append(']').toString();
    }

    private static String toJsonScalar(Object value) {
        if (value == null) {
            return "null";
        }
        if (value instanceof Boolean || value instanceof Number) {
            return value.toString();
        }
        return '"' + escapeJsonString(value.toString()) + '"';
    }

    private static String escapeJsonString(String s) {
        StringBuilder sb = new StringBuilder(s.length() + 8);
        for (int i = 0; i < s.length(); i++) {
            char c = s.charAt(i);
            switch (c) {
                case '"' -> sb.append("\\\"");
                case '\\' -> sb.append("\\\\");
                case '\n' -> sb.append("\\n");
                case '\r' -> sb.append("\\r");
                case '\t' -> sb.append("\\t");
                default -> {
                    if (c < 0x20) {
                        sb.append(String.format("\\u%04x", (int) c));
                    } else {
                        sb.append(c);
                    }
                }
            }
        }
        return sb.toString();
    }
}
"#;

// ---------------------------------------------------------------------
// Message records
// ---------------------------------------------------------------------

/// Every distinct message type referenced anywhere in `doc`, keyed by the
/// Java type name it will generate as (same naming rule as
/// `RustCodeGenerator::ref_to_rust_type`: PascalCase of the reference
/// pointer's last segment, which for an Internal Reference is the
/// `components.messages.<id>` id). Unlike
/// `RustCodeGenerator::collect_internal_message_ids`, this collects BOTH
/// External and Internal References — the Java target has no equivalent of
/// Rust's assumed pre-existing `{alias}::messages` crate, so every message
/// actually used gets a real generated record, resolved through `registry`
/// either way (see `AsyncApiRegistry::resolve_message`).
fn collect_message_refs(doc: &EtlDocument) -> BTreeMap<String, MessageRef> {
    let mut refs: BTreeMap<String, MessageRef> = BTreeMap::new();
    let note = |m: &MessageRef, refs: &mut BTreeMap<String, MessageRef>| {
        let name = ref_to_type_name(m);
        refs.entry(name).or_insert_with(|| m.clone());
    };

    for tree in doc.event_trees.values() {
        note(&tree.initiating_event.message, &mut refs);
        for node in tree.nodes.values() {
            match node {
                Node::Operation(op) => {
                    if let Some(ref m) = op.emits {
                        note(m, &mut refs);
                    }
                }
                Node::Consequence(cons) => {
                    if let Some(ref m) = cons.message {
                        note(m, &mut refs);
                    }
                }
                _ => {}
            }
        }
    }

    if let Some(ref fault_trees) = doc.fault_trees {
        for ft in fault_trees.values() {
            if let Some(ref m) = ft.top_event.message {
                note(m, &mut refs);
            }
            for be in ft.basic_events.values() {
                if let Some(ref m) = be.message {
                    note(m, &mut refs);
                }
            }
        }
    }

    refs
}

fn ref_to_type_name(msg_ref: &MessageRef) -> String {
    let pointer = match msg_ref {
        MessageRef::External(r) => &r.pointer,
        MessageRef::Internal(r) => &r.pointer,
    };
    extract_last_segment(pointer)
}

fn extract_last_segment(pointer: &str) -> String {
    let last = pointer.rsplit('/').next().unwrap_or("Unknown");
    to_pascal_case(last)
}

/// Generates the `<package>/<TypeName>.java` file for one resolved AsyncAPI
/// Message Object value (`{name?, payload, headers?}`, the same envelope
/// shape `AsyncApiRegistry::resolve_message` normalizes both reference
/// kinds into). The envelope record (`public`, matching the filename) is
/// emitted last; the payload record and any nested-object records it needs
/// are emitted above it as package-private top-level records in the same
/// file — Java permits any number of non-public top-level types per file,
/// only one `public` type, which must match the filename.
fn generate_message_file(package: &str, type_name: &str, message: &serde_json::Value) -> String {
    let payload_type_name = format!("{type_name}Payload");
    let payload_schema = message.get("payload").cloned().unwrap_or(serde_json::Value::Null);

    let mut nested = String::new();
    let payload_java_type = schema_to_java_type(&payload_type_name, &payload_schema, &mut nested, false);

    let mut out = String::new();
    out.push_str(&format!("package {package};\n\n"));
    out.push_str("import java.util.List;\nimport java.util.Map;\n\n");
    out.push_str("// AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY\n\n");
    out.push_str(&nested);
    out.push_str(&format!(
        "public record {type_name}({payload_java_type} payload, Map<String, Object> headers) {{}}\n"
    ));
    out
}

/// Java analog of `RustCodeGenerator::generate_struct_from_schema`: emits a
/// package-private `record <name>(...)` for an object-typed JSON Schema. A
/// schema with no `properties` (or a non-object schema) falls back to
/// `Map<String, Object>` at the call site instead — unlike the Rust target's
/// always-present newtype wrapper, Java has no ceremony-free equivalent, and
/// a bare `Map` is idiomatic enough for a schema too loose to model.
fn generate_record_from_schema(name: &str, schema: &serde_json::Value, output: &mut String) {
    let Some(properties) = schema.get("properties").and_then(|p| p.as_object()) else {
        return;
    };

    let required: std::collections::BTreeSet<&str> = schema
        .get("required")
        .and_then(|r| r.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();

    let mut nested = String::new();
    let mut components = Vec::new();

    for (field_name, field_schema) in properties {
        let camel = to_camel_case(field_name);
        let nested_type_name = format!("{name}{}", to_pascal_case(field_name));
        let is_required = required.contains(field_name.as_str());
        let java_type = schema_to_java_type(&nested_type_name, field_schema, &mut nested, !is_required);
        components.push(format!("{java_type} {camel}"));
    }

    output.push_str(&nested);
    output.push_str(&format!("record {name}({}) {{}}\n\n", components.join(", ")));
}

/// Java analog of `RustCodeGenerator::schema_to_rust_type`. `nullable`
/// requests a boxed wrapper for a JSON Schema primitive (Java has no
/// primitive `Option<T>`-equivalent; a non-required field becomes a
/// nullable boxed type instead, the idiomatic Java equivalent).
fn schema_to_java_type(
    candidate_name: &str,
    schema: &serde_json::Value,
    nested: &mut String,
    nullable: bool,
) -> String {
    match schema.get("type").and_then(|t| t.as_str()) {
        Some("string") => "String".to_string(),
        Some("integer") => if nullable { "Long" } else { "long" }.to_string(),
        Some("number") => if nullable { "Double" } else { "double" }.to_string(),
        Some("boolean") => if nullable { "Boolean" } else { "boolean" }.to_string(),
        Some("array") => {
            let item_type = match schema.get("items") {
                Some(items_schema) => schema_to_java_type(
                    &format!("{candidate_name}Item"),
                    items_schema,
                    nested,
                    true,
                ),
                None => "Object".to_string(),
            };
            format!("List<{item_type}>")
        }
        Some("object") => {
            if schema.get("properties").is_some() {
                generate_record_from_schema(candidate_name, schema, nested);
                candidate_name.to_string()
            } else {
                "Map<String, Object>".to_string()
            }
        }
        _ => "Object".to_string(),
    }
}

// ---------------------------------------------------------------------
// Handler interfaces (one per event tree — the user-editable boundary)
// ---------------------------------------------------------------------

/// `handler:` ids (core spec Section 5.5.2) used by every Operation node in
/// `tree`, deduplicated, in first-seen order. Every call site for a given
/// tree passes the same message type (the tree's initiating event message
/// type — codegen never rebinds `message` to a different type within a
/// tree), so one interface per tree, not per document, avoids any signature
/// collision between trees that happen to reuse a handler id with a
/// different message type.
fn collect_handler_methods(tree: &EventTree) -> Vec<String> {
    let mut seen = std::collections::BTreeSet::new();
    let mut methods = Vec::new();
    for node in tree.nodes.values() {
        if let Node::Operation(op) = node {
            let name = to_camel_case(&op.handler);
            if seen.insert(name.clone()) {
                methods.push(name);
            }
        }
    }
    methods
}

fn generate_handlers_interface(
    package: &str,
    iface_name: &str,
    msg_type: &str,
    methods: &[String],
) -> String {
    let mut out = String::new();
    out.push_str(&format!("package {package};\n\n"));
    out.push_str("import etdl.runtime.WorkflowError;\n\n");
    out.push_str("// AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY\n");
    out.push_str("// Implement this interface in your own (never regenerated) class; the\n");
    out.push_str("// generated orchestration in <Stem>Workflow calls through it and never the\n");
    out.push_str("// other way around.\n");
    out.push_str(&format!("public interface {iface_name} {{\n"));
    for method in methods {
        out.push_str(&format!(
            "    Object {method}({msg_type} message) throws WorkflowError;\n"
        ));
    }
    out.push_str("}\n");
    out
}

// ---------------------------------------------------------------------
// Workflow orchestration class
// ---------------------------------------------------------------------

fn generate_workflow_class(
    doc: &EtlDocument,
    package: &str,
    class_stem: &str,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Result<String, String> {
    let mut out = String::new();
    out.push_str(&format!("package {package};\n\n"));
    out.push_str("import etdl.runtime.*;\nimport java.util.List;\nimport java.util.Map;\n\n");
    out.push_str(&format!(
        "// AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY\n\npublic final class {class_stem}Workflow {{\n"
    ));
    out.push_str(&format!("    private {class_stem}Workflow() {{}}\n\n"));

    out.push_str(&generate_fault_tree_constants(doc, fault_tree_probs));

    for tree in doc.event_trees.values() {
        let method = generate_event_tree_method(doc, tree, fault_tree_probs)?;
        out.push_str(&method);
        out.push('\n');
    }

    out.push_str("}\n");
    Ok(out)
}

fn generate_fault_tree_constants(doc: &EtlDocument, fault_tree_probs: &BTreeMap<String, f64>) -> String {
    let mut output = String::new();
    for tree in doc.event_trees.values() {
        for (node_id, node) in &tree.nodes {
            if let Node::Operation(op) = node {
                if let Some(ref ps) = op.on_failure_probability_source {
                    if let Some((ft_id, prob)) = find_fault_tree_prob(ps, fault_tree_probs) {
                        output.push_str(&format!(
                            "    // Computed from faultTrees.{ft_id}.topEvent at build time (Section 5.16)\n"
                        ));
                        let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
                        output.push_str(&format!(
                            "    public static final double {const_name} = {prob:.6};\n\n"
                        ));
                    }
                }
            }
        }
    }
    output
}

fn find_fault_tree_prob(
    ps: &etdl_parser::ast::InternalRef,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Option<(String, f64)> {
    let ft_id = extract_ft_id(&ps.pointer);
    fault_tree_probs.get(&ft_id).map(|&v| (ft_id.clone(), v))
}

fn extract_ft_id(pointer: &str) -> String {
    pointer
        .trim_start_matches("#/faultTrees/")
        .trim_end_matches("/topEvent")
        .to_string()
}

fn generate_event_tree_method(
    doc: &EtlDocument,
    tree: &EventTree,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Result<String, String> {
    let mut out = String::new();

    let tree_id = tree_id_of(doc, tree);
    let tree_class = to_pascal_case(&sanitize_ident(tree_id));
    let handlers_iface = format!("{tree_class}Handlers");
    let method_name = format!("handle{}", to_pascal_case(&sanitize_ident(&tree.initiating_event.id)));
    let msg_type = ref_to_type_name(&tree.initiating_event.message);

    out.push_str(&format!(
        "    public static void {method_name}({msg_type} message, Publisher publisher, {handlers_iface} handlers) throws WorkflowError {{\n"
    ));

    let first_barrier = find_first_barrier(tree);
    let monitor_name = first_barrier.map(to_camel_case).unwrap_or_else(|| "monitor".to_string());
    // `BranchMonitor` owns a native handle (see `etdl.runtime.BranchMonitor`)
    // — try-with-resources guarantees `etdl_branch_monitor_free` runs when
    // this event-tree invocation completes, success or failure, exactly
    // matching the one-monitor-per-invocation lifetime generated Rust code
    // already uses.
    out.push_str(&format!(
        "        try (BranchMonitor {monitor_name} = new BranchMonitor(\"{}\")) {{\n",
        first_barrier.unwrap_or("root")
    ));

    let start_node_id = &tree.initiating_event.next;
    let body = generate_node_code(doc, tree, start_node_id, 3, fault_tree_probs, &monitor_name)?;
    out.push_str(&body);

    out.push_str("        }\n");
    out.push_str("    }\n");
    Ok(out)
}

fn find_first_barrier(tree: &EventTree) -> Option<&str> {
    let mut current = &tree.initiating_event.next;
    loop {
        match tree.nodes.get(current.as_str()) {
            Some(Node::Barrier(_)) => return Some(current.as_str()),
            Some(Node::Operation(op)) => current = &op.next,
            Some(Node::Consequence(_)) => return None,
            None => return None,
        }
    }
}

fn generate_node_code(
    doc: &EtlDocument,
    tree: &EventTree,
    node_id: &str,
    depth: usize,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let node = tree
        .nodes
        .get(node_id)
        .ok_or_else(|| format!("node '{node_id}' not found"))?;

    match node {
        Node::Barrier(barrier) => {
            generate_barrier_code(tree, node_id, barrier, depth, doc, fault_tree_probs, monitor_name)
        }
        Node::Operation(op) => {
            generate_operation_code(tree, node_id, op, depth, doc, fault_tree_probs, monitor_name)
        }
        Node::Consequence(cons) => generate_consequence_code(cons, depth),
    }
}

fn generate_barrier_code(
    tree: &EventTree,
    node_id: &str,
    barrier: &etdl_parser::ast::Barrier,
    depth: usize,
    doc: &EtlDocument,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    for (i, branch) in barrier.branches.iter().enumerate() {
        let is_default = branch.condition == Condition::Default;
        if i == 0 {
            if is_default {
                emit_branch_record(&indent, monitor_name, branch, node_id, fault_tree_probs, &mut output);
                let body = generate_node_code(doc, tree, &branch.next, depth + 1, fault_tree_probs, monitor_name)?;
                output.push_str(&body);
            } else {
                let cond = condition_to_java_code(&branch.condition);
                output.push_str(&format!("{indent}if ({cond}) {{\n"));
                emit_branch_record(&format!("{indent}    "), monitor_name, branch, node_id, fault_tree_probs, &mut output);
                let body = generate_node_code(doc, tree, &branch.next, depth + 1, fault_tree_probs, monitor_name)?;
                output.push_str(&body);
                output.push_str(&format!("{indent}}}"));
            }
        } else if is_default {
            output.push_str(" else {\n");
            emit_branch_record(&format!("{indent}    "), monitor_name, branch, node_id, fault_tree_probs, &mut output);
            let body = generate_node_code(doc, tree, &branch.next, depth + 1, fault_tree_probs, monitor_name)?;
            output.push_str(&body);
            output.push_str(&format!("{indent}}}\n"));
        } else {
            let cond = condition_to_java_code(&branch.condition);
            output.push_str(&format!(" else if ({cond}) {{\n"));
            emit_branch_record(&format!("{indent}    "), monitor_name, branch, node_id, fault_tree_probs, &mut output);
            let body = generate_node_code(doc, tree, &branch.next, depth + 1, fault_tree_probs, monitor_name)?;
            output.push_str(&body);
            output.push_str(&format!("{indent}}}"));
        }
    }

    if !barrier.branches.is_empty() && barrier.branches.last().unwrap().condition != Condition::Default {
        output.push('\n');
    }

    Ok(output)
}

fn emit_branch_record(
    indent: &str,
    monitor_name: &str,
    branch: &etdl_parser::ast::Branch,
    node_id: &str,
    fault_tree_probs: &BTreeMap<String, f64>,
    output: &mut String,
) {
    if let Some(p) = get_branch_prob(branch, node_id, fault_tree_probs) {
        output.push_str(&format!(
            "{indent}{monitor_name}.recordBranch(\"{}\", {p:.6});\n",
            branch.outcome
        ));
    }
}

fn get_branch_prob(
    branch: &etdl_parser::ast::Branch,
    _node_id: &str,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Option<f64> {
    if let Some(ref ps) = branch.probability_source {
        let ft_id = extract_ft_id(&ps.pointer);
        return fault_tree_probs.get(&ft_id).copied();
    }
    branch.effective_probability()
}

fn generate_operation_code(
    tree: &EventTree,
    node_id: &str,
    op: &etdl_parser::ast::Operation,
    depth: usize,
    doc: &EtlDocument,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    let handler_name = to_camel_case(&op.handler);
    let timeout = op.timeout_ms.unwrap_or(5000);

    if let Some(ref retry) = op.retry_policy {
        let strategy = match retry.backoff_strategy.as_ref().unwrap_or(&AstBackoffStrategy::Fixed) {
            AstBackoffStrategy::Exponential => "BackoffStrategy.EXPONENTIAL",
            AstBackoffStrategy::Fixed => "BackoffStrategy.FIXED",
        };
        // `RetryPolicy` owns a native handle — try-with-resources frees it
        // (`etdl_retry_policy_free`) whether this attempt sequence succeeds
        // or every attempt fails.
        output.push_str(&format!(
            "{indent}try (RetryPolicy retry = new RetryPolicy({}, {}, {strategy})) {{\n",
            retry.max_attempts, retry.backoff_ms
        ));
        output.push_str(&format!(
            "{indent}    Object _result = retry.execute(() -> handlers.{handler_name}(message), {timeout}L);\n"
        ));
    } else {
        output.push_str(&format!("{indent}try {{\n"));
        output.push_str(&format!(
            "{indent}    Object _result = handlers.{handler_name}(message);\n"
        ));
    }

    if op.on_failure.is_some() {
        if let Some(ref ps) = op.on_failure_probability_source {
            let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
            let prob_exists = find_fault_tree_prob(ps, fault_tree_probs).is_some();
            if prob_exists {
                output.push_str(&format!(
                    "{indent}    {monitor_name}.recordSuccess(\"{node_id}\", {const_name});\n"
                ));
            } else {
                output.push_str(&format!(
                    "{indent}    {monitor_name}.recordSuccess(\"{node_id}\", null);\n"
                ));
            }
        }
    }

    let next_node = tree.nodes.get(&op.next);
    match next_node {
        Some(Node::Consequence(cons)) => {
            emit_send(cons, "_result", &format!("{indent}    "), &mut output);
        }
        Some(_) => {
            let body = generate_node_code(doc, tree, &op.next, depth + 1, fault_tree_probs, monitor_name)?;
            output.push_str(&body);
        }
        None => {}
    }

    output.push_str(&format!("{indent}}} catch (WorkflowError err) {{\n"));

    if let Some(ref ps) = op.on_failure_probability_source {
        let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
        let prob_exists = find_fault_tree_prob(ps, fault_tree_probs).is_some();
        if prob_exists {
            output.push_str(&format!(
                "{indent}    {monitor_name}.recordFailure(\"{node_id}\", err, {const_name});\n"
            ));
        } else {
            output.push_str(&format!(
                "{indent}    {monitor_name}.recordFailure(\"{node_id}\", err, null);\n"
            ));
        }
    } else {
        output.push_str(&format!(
            "{indent}    {monitor_name}.recordFailure(\"{node_id}\", err, null);\n"
        ));
    }

    if let Some(ref on_failure_id) = op.on_failure {
        match tree.nodes.get(on_failure_id) {
            Some(Node::Consequence(cons)) => {
                emit_send(cons, "message", &format!("{indent}    "), &mut output);
            }
            _ => {
                let body = generate_node_code(doc, tree, on_failure_id, depth + 1, fault_tree_probs, monitor_name)?;
                output.push_str(&body);
            }
        }
    } else {
        output.push_str(&format!(
            "{indent}    throw new WorkflowError(String.valueOf(err.getMessage()), err);\n"
        ));
    }

    output.push_str(&format!("{indent}}}\n"));

    Ok(output)
}

fn emit_send(cons: &etdl_parser::ast::Consequence, payload_expr: &str, indent: &str, output: &mut String) {
    if let etdl_parser::ast::ConsequenceOperation::Send = cons.consequence_operation {
        if let Some(ref channel_ref) = cons.channel {
            let channel_name = channel_ref_name(channel_ref);
            output.push_str(&format!(
                "{indent}publisher.publish(\"{channel_name}\", {payload_expr});\n"
            ));
        }
    }
}

fn generate_consequence_code(cons: &etdl_parser::ast::Consequence, depth: usize) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    if let etdl_parser::ast::ConsequenceOperation::Send = cons.consequence_operation {
        if let Some(ref channel_ref) = cons.channel {
            let channel_name = channel_ref_name(channel_ref);
            output.push_str(&format!(
                "{indent}publisher.publish(\"{channel_name}\", message);\n"
            ));
        }
    }

    Ok(output)
}

fn channel_ref_name(channel_ref: &ChannelRef) -> String {
    match channel_ref {
        ChannelRef::External(ext_ref) => extract_last_segment(&ext_ref.pointer),
        ChannelRef::Bare(name) => name.clone(),
    }
}

// ---------------------------------------------------------------------
// ECEL condition -> Java
// ---------------------------------------------------------------------

fn condition_to_java_code(condition: &Condition) -> String {
    match condition {
        Condition::Default => "true".to_string(),
        Condition::Expr(expr) => render_bool_expr(expr),
    }
}

/// `boolean-expr` (spec §6.2): any combination of `comparison`/
/// `quantifier-expr`/`defined-expr` joined by `&&`/`||`/unary `!`. A bare
/// `Comparison` (no combinator) is the pre-existing, common case; `And`/
/// `Or`/`Not` map directly onto Java's own operators.
fn render_bool_expr(expr: &ecel::BoolExpr) -> String {
    use ecel::BoolExpr as B;
    match expr {
        B::And(a, b) => format!("({}) && ({})", render_bool_expr(a), render_bool_expr(b)),
        B::Or(a, b) => format!("({}) || ({})", render_bool_expr(a), render_bool_expr(b)),
        B::Not(a) => format!("!({})", render_bool_expr(a)),
        B::Comparison(cmp) => render_comparison(cmp),
        B::Quantifier(q) => render_quantifier(q),
        B::Defined(path) => render_defined(path),
    }
}

fn render_comparison(cmp: &ecel::Comparison) -> String {
    use ecel::Comparator as C;
    match cmp.op {
        C::In => {
            let left = path_or_literal(&cmp.left);
            let right = array_or_literal(&cmp.right);
            format!("etdl.runtime.Condition.contains({right}, {left})")
        }
        C::Matches => {
            let left = match as_path(&cmp.left) {
                Some(_) => operand_to_path_expr(&cmp.left),
                None => operand_to_val_str(&cmp.left),
            };
            let right = literal_to_val_str(&literal_of(&cmp.right));
            format!("etdl.runtime.Condition.matches({left}, {right})")
        }
        _ => {
            let (left_path, has_wildcard) = match as_path(&cmp.left) {
                Some(p) => build_path_expression(p),
                None => (PathParts { path_prefix: String::new(), remaining_path: String::new() }, false),
            };
            let right = operand_to_val_str(&cmp.right);
            let op = comparator_str(&cmp.op);

            if has_wildcard {
                format!(
                    "{}.stream().allMatch(item -> item{} {} {})",
                    left_path.path_prefix, left_path.remaining_path, op, right
                )
            } else {
                let l = if left_path.path_prefix.is_empty() && left_path.remaining_path.is_empty() {
                    operand_to_val_str(&cmp.left)
                } else {
                    left_path.path_prefix + &left_path.remaining_path
                };
                let r = if right.is_empty() { operand_to_path_expr(&cmp.right) } else { right };
                format!("{l} {op} {r}")
            }
        }
    }
}

/// `quantifier-expr` (spec §6.4): `any`/`all` over a wildcarded path's
/// matching elements. Reuses the same wildcard-split machinery a bare
/// comparison's implicit-`all` case already uses (`render_comparison`),
/// parameterized on `any`/`all` — Java's `Stream.anyMatch`/`.allMatch` map
/// directly onto the two `QuantifierKind` values. A validly type-checked
/// document always has a wildcarded payload path here (spec §6.4); if not,
/// degrade to the plain comparison rather than emit nonsense.
fn render_quantifier(q: &ecel::QuantifierExpr) -> String {
    if let Some(path_expr) = as_path(&q.comparison.left) {
        if path_has_wildcard(path_expr) && !is_headers_path(path_expr) {
            let (parts, _has_wildcard) = build_path_expression(path_expr);
            let right = operand_to_val_str(&q.comparison.right);
            let op = comparator_str(&q.comparison.op);
            let method = match q.kind {
                ecel::QuantifierKind::Any => "anyMatch",
                ecel::QuantifierKind::All => "allMatch",
            };
            return format!(
                "{}.stream().{}(item -> item{} {} {})",
                parts.path_prefix, method, parts.remaining_path, op, right
            );
        }
    }
    render_comparison(&q.comparison)
}

/// `defined-expr` (spec §6.2): whether a path both exists and is non-null.
/// `headers` is a raw `Map<String, Object>` on the generated message
/// record, so a missing key and an absent field are the same
/// `.get(...) == null` check; a payload path is a typed record accessor
/// chain, so only null-ness is checkable (a required, non-nullable field
/// is always "defined" by construction — `defined()` on one degrades to
/// `true`, matching how a required Rust struct field has no `Option`
/// wrapper to be absent through either).
fn render_defined(path: &ecel::PathExpr) -> String {
    if is_headers_path(path) {
        let mut expr = String::from("message.headers()");
        for seg in path.segments.iter().skip(2) {
            let key = match seg {
                ecel::PathSegment::Field(name) | ecel::PathSegment::QuotedKey(name) => {
                    format!("\"{name}\"")
                }
                ecel::PathSegment::Index(idx) => idx.to_string(),
                ecel::PathSegment::Wildcard => continue,
            };
            expr = format!("(({expr}) instanceof java.util.Map<?, ?> __etdlM ? __etdlM.get({key}) : null)");
        }
        format!("(({expr}) != null)")
    } else {
        format!("(({}) != null)", render_path_expr(path))
    }
}

fn path_or_literal(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(_) => operand_to_path_expr(operand),
        None => operand_to_val_str(operand),
    }
}

fn array_or_literal(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(_) => operand_to_path_expr(operand),
        None => operand_to_val_str(operand),
    }
}

fn literal_of(operand: &ecel::Operand) -> ecel::Literal {
    match operand {
        ecel::Operand::Literal(lit) => lit.clone(),
        ecel::Operand::Value(_) => ecel::Literal::String(String::new()),
    }
}

/// `Some(path)` if `operand` is a bare path (spec §6.2's `value-expr ::=
/// path-expr`), independent of whether it's used at the top level of a
/// comparison (where a path gets wildcard-quantification treatment via
/// `build_path_expression`) or nested inside arithmetic/a function call
/// (where `render_value_expr`'s own `Path` arm renders it directly).
fn as_path(operand: &ecel::Operand) -> Option<&ecel::PathExpr> {
    match operand {
        ecel::Operand::Value(ecel::ValueExpr::Path(p)) => Some(p),
        _ => None,
    }
}

fn path_has_wildcard(path: &ecel::PathExpr) -> bool {
    path.segments.iter().any(|s| matches!(s, ecel::PathSegment::Wildcard))
}

/// Whether `path` is rooted at `message.headers` (spec §6.3's second root)
/// rather than `message.payload` — decided purely syntactically (the
/// segment right after `message`), never requiring schema access, matching
/// `RustCodeGenerator::path_root`'s own discipline.
fn is_headers_path(path: &ecel::PathExpr) -> bool {
    matches!(path.segments.get(1), Some(ecel::PathSegment::Field(name)) if name == "headers")
}

struct PathParts {
    path_prefix: String,
    remaining_path: String,
}

/// Java analog of `RustCodeGenerator::build_path_expression`. Segments
/// render as record-accessor calls (`.field()`) rather than field access,
/// since generated message types are Java records; `Index`/`QuotedKey`
/// segments render as `.get(...)` calls (the fallback types for
/// under-specified schema are `List<T>`/`Map<String, Object>`), where
/// Rust's structurally different fallback (`Vec`/`serde_json::Value`) could
/// use bracket indexing directly.
fn build_path_expression(path_expr: &ecel::PathExpr) -> (PathParts, bool) {
    let segments = &path_expr.segments;
    let mut pre_wildcard = Vec::new();
    let mut post_wildcard = Vec::new();
    let mut has_wildcard = false;

    for (i, seg) in segments.iter().enumerate() {
        if i == 0 {
            continue;
        }
        if has_wildcard {
            post_wildcard.push(seg.clone());
        } else if matches!(seg, ecel::PathSegment::Wildcard) {
            has_wildcard = true;
        } else {
            pre_wildcard.push(seg.clone());
        }
    }

    let mut prefix = String::from("message");
    for seg in &pre_wildcard {
        append_segment(&mut prefix, seg);
    }

    let mut suffix = String::new();
    for seg in &post_wildcard {
        append_segment(&mut suffix, seg);
    }

    (
        PathParts { path_prefix: prefix, remaining_path: suffix },
        has_wildcard,
    )
}

fn append_segment(out: &mut String, seg: &ecel::PathSegment) {
    match seg {
        ecel::PathSegment::Field(name) => {
            out.push('.');
            out.push_str(&to_camel_case(name));
            out.push_str("()");
        }
        ecel::PathSegment::Index(idx) => {
            out.push_str(&format!(".get({idx})"));
        }
        ecel::PathSegment::QuotedKey(name) => {
            out.push_str(&format!(".get(\"{name}\")"));
        }
        _ => {}
    }
}

/// Core path-segment rendering, shared by the top-level
/// `operand_to_path_expr` (bare `Operand::Value(Path(_))`, used by the
/// wildcard-comparison machinery above) and `render_value_expr`'s `Path`
/// arm (a path nested inside an arithmetic expression or function-call
/// argument, e.g. `length(message.payload.name)`).
fn render_path_expr(path: &ecel::PathExpr) -> String {
    let mut out = String::from("message");
    for seg in path.segments.iter().skip(1) {
        if matches!(seg, ecel::PathSegment::Wildcard) {
            continue;
        }
        append_segment(&mut out, seg);
    }
    out
}

fn operand_to_path_expr(operand: &ecel::Operand) -> String {
    match operand {
        ecel::Operand::Value(ecel::ValueExpr::Path(path)) => render_path_expr(path),
        _ => String::new(),
    }
}

fn operand_to_val_str(operand: &ecel::Operand) -> String {
    match operand {
        // A bare path renders empty here on purpose: `condition_to_java_code`
        // tries `build_path_expression`/`operand_to_path_expr` first for a
        // path operand (to get wildcard-quantification right) and only
        // falls back to this function for the non-path cases below.
        ecel::Operand::Value(ecel::ValueExpr::Path(_)) => String::new(),
        ecel::Operand::Value(v) => render_value_expr(v),
        ecel::Operand::Literal(lit) => literal_to_val_str(lit),
    }
}

/// Renders a `value-expr` (spec §6.2: path/number/func-call, optionally
/// combined with `+`/`-`/`*`/`/`) as a Java expression. Generated message
/// types are Java records, so a numeric payload field's accessor already
/// returns a primitive `double`/boxed `Double` — no extra coercion needed
/// before using it in arithmetic, unlike Rust's `as_f64_expr` (which has to
/// bridge `Option<T>`/integer-vs-float distinctions Java's schema mapping
/// (`docs`/`Payload::to_java_type`) doesn't have).
fn render_value_expr(expr: &ecel::ValueExpr) -> String {
    use ecel::{FuncName as F, ValueExpr as V};
    match expr {
        V::Path(path) => render_path_expr(path),
        V::Number(n) => n.to_string(),
        V::Call(func, arg) => {
            let arg_code = render_value_expr(arg);
            match func {
                // `length()` (spec §6.5.1) is the codepoint/element count of
                // a string or array; unlike Rust's static, schema-lookup-based
                // decision (`render_length_call`), this dispatches at
                // runtime via `etdl.runtime.Condition.length(Object)` — a
                // payload accessor's concrete static type (e.g. `String`)
                // makes an inline `instanceof List<?>` a compile error
                // (provably-unrelated types), not just a false branch; see
                // that helper's own doc comment.
                F::Length => format!("etdl.runtime.Condition.length({arg_code})"),
                F::Abs => format!("Math.abs({arg_code})"),
                F::Lower => format!("({arg_code}).toLowerCase()"),
                F::Upper => format!("({arg_code}).toUpperCase()"),
            }
        }
        V::Add(a, b) => format!("(({}) + ({}))", render_value_expr(a), render_value_expr(b)),
        V::Sub(a, b) => format!("(({}) - ({}))", render_value_expr(a), render_value_expr(b)),
        V::Mul(a, b) => format!("(({}) * ({}))", render_value_expr(a), render_value_expr(b)),
        V::Div(a, b) => format!("(({}) / ({}))", render_value_expr(a), render_value_expr(b)),
    }
}

fn literal_to_val_str(lit: &ecel::Literal) -> String {
    match lit {
        ecel::Literal::Number(n) => n.to_string(),
        ecel::Literal::String(s) => format!("\"{s}\""),
        ecel::Literal::Bool(b) => b.to_string(),
        ecel::Literal::Null => "null".to_string(),
        ecel::Literal::Array(items) => {
            let inner: Vec<String> = items.iter().map(literal_to_val_str).collect();
            format!("List.of({})", inner.join(", "))
        }
    }
}

fn comparator_str(op: &ecel::Comparator) -> &str {
    match op {
        ecel::Comparator::Eq => "==",
        ecel::Comparator::Neq => "!=",
        ecel::Comparator::Gte => ">=",
        ecel::Comparator::Lte => "<=",
        ecel::Comparator::Gt => ">",
        ecel::Comparator::Lt => "<",
        ecel::Comparator::In => "in",
        ecel::Comparator::Matches => "matches",
    }
}

// ---------------------------------------------------------------------
// Naming helpers
// ---------------------------------------------------------------------

fn sanitize_package(domain: &str) -> String {
    let segments: Vec<String> = domain
        .split(|c: char| !c.is_ascii_alphanumeric())
        .filter(|s| !s.is_empty())
        .map(|s| {
            let lower = s.to_ascii_lowercase();
            if lower.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(false) {
                format!("_{lower}")
            } else {
                lower
            }
        })
        .collect();

    if segments.is_empty() {
        "etdl.generated".to_string()
    } else {
        segments.join(".")
    }
}

fn sanitize_ident(s: &str) -> String {
    let cleaned: String = s
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect();
    if cleaned.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(true) {
        format!("_{cleaned}")
    } else {
        cleaned
    }
}

fn to_snake_case(s: &str) -> String {
    let mut result = String::new();
    for (i, c) in s.chars().enumerate() {
        if c.is_uppercase() {
            if i > 0 {
                result.push('_');
            }
            result.push(c.to_lowercase().next().unwrap());
        } else {
            result.push(c);
        }
    }
    result
}

fn to_upper_snake(s: &str) -> String {
    to_snake_case(s).to_uppercase()
}

fn to_pascal_case(s: &str) -> String {
    let mut result = String::new();
    let mut capitalize = true;
    for c in s.chars() {
        if c == '_' || c == '-' || c == ' ' {
            capitalize = true;
        } else if capitalize {
            result.extend(c.to_uppercase());
            capitalize = false;
        } else {
            result.push(c);
        }
    }
    result
}

fn to_camel_case(s: &str) -> String {
    let pascal = to_pascal_case(s);
    let mut chars = pascal.chars();
    match chars.next() {
        Some(c) => c.to_lowercase().collect::<String>() + chars.as_str(),
        None => String::new(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn package_sanitizes_domain() {
        assert_eq!(sanitize_package("com.example.orders"), "com.example.orders");
        assert_eq!(sanitize_package(""), "etdl.generated");
        assert_eq!(sanitize_package("Order-Fulfillment"), "order.fulfillment");
    }

    #[test]
    fn camel_and_pascal_case() {
        assert_eq!(to_pascal_case("order_placed"), "OrderPlaced");
        assert_eq!(to_camel_case("order_placed"), "orderPlaced");
        assert_eq!(to_camel_case("HandlerName"), "handlerName");
    }

    #[test]
    fn in_operator_lowers_to_condition_contains() {
        let cond = ecel::parse_condition(
            "message.payload.status in [\"PAID\", \"AUTHORIZED\"]",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("etdl.runtime.Condition.contains"), "got: {code}");
        assert!(code.contains("\"PAID\""));
    }

    #[test]
    fn matches_operator_lowers_to_condition_matches() {
        let cond = ecel::parse_condition(
            "message.payload.reference matches \"^ORD-[0-9]{8}$\"",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("etdl.runtime.Condition.matches"), "got: {code}");
    }

    #[test]
    fn comparison_uses_record_accessors() {
        let cond = ecel::parse_condition("message.payload.amount >= 10000").unwrap();
        let code = condition_to_java_code(&cond);
        assert_eq!(code, "message.payload().amount() >= 10000");
    }

    #[test]
    fn boolean_and_or_not_combine_comparisons() {
        let cond = ecel::parse_condition(
            "message.payload.amount >= 10000 && !(message.payload.status == \"VOID\")",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("&&"), "got: {code}");
        assert!(code.contains("!("), "got: {code}");

        let cond = ecel::parse_condition(
            "message.payload.amount >= 10000 || message.payload.priority == true",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("||"), "got: {code}");
    }

    #[test]
    fn arithmetic_and_function_calls_render_as_java_expressions() {
        let cond = ecel::parse_condition("message.payload.amount + 5 > 100").unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("message.payload().amount()"), "got: {code}");
        assert!(code.contains('+'), "got: {code}");

        let cond = ecel::parse_condition("abs(message.payload.amount) > 100").unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("Math.abs("), "got: {code}");

        let cond = ecel::parse_condition("length(message.payload.reference) > 5").unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("etdl.runtime.Condition.length("), "got: {code}");
    }

    #[test]
    fn quantifier_lowers_to_stream_any_or_all_match() {
        let cond = ecel::parse_condition(
            "any(message.payload.items, message.payload.items[*].qty > 0)",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains(".stream().anyMatch("), "got: {code}");

        let cond = ecel::parse_condition(
            "all(message.payload.items, message.payload.items[*].qty > 0)",
        )
        .unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains(".stream().allMatch("), "got: {code}");
    }

    #[test]
    fn defined_checks_payload_and_headers_paths() {
        let cond = ecel::parse_condition("defined(message.payload.reference)").unwrap();
        let code = condition_to_java_code(&cond);
        assert_eq!(code, "((message.payload().reference()) != null)");

        let cond = ecel::parse_condition("defined(message.headers.traceId)").unwrap();
        let code = condition_to_java_code(&cond);
        assert!(code.contains("message.headers()"), "got: {code}");
        assert!(code.contains("!= null"), "got: {code}");
    }
}