Idempotency and Correlation

        // Every workflow instance has a globally unique, stable identifier
        String instanceId = ctx.instanceData().id(); // e.g. "01K9GDCXJVN89V0N4CWVG40R7C"

@POST
@Path("/payments")
public Response processPayment(@HeaderParam("X-Order-Id") String orderId,
    PaymentRequest request) {
// Check if we've already processed this order
    if (paymentStore.exists(orderId)) {
        // Return the previously stored result — no double-charge
        return Response.ok(paymentStore.get(orderId)).build();
    }

    // Process the payment (this runs at most once per order)
    PaymentResult result = paymentGateway.charge(request);

    // Store the result keyed by the business key
    paymentStore.save(orderId, result);

    return Response.ok(result).build();
}

import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;

public class PaymentWorkflow extends Flow {

    public Workflow descriptor() {
        return workflow("paymentWorkflow")
            .tasks(
                call("processPayment", http("/payments")
                    .method("POST")
                    .header("X-Order-Id", "${.order.id}")
                    .body(".paymentRequest"))
            )
            .build();
    }
}

package org.acme;

import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;

import jakarta.enterprise.context.ApplicationScoped;

import io.quarkiverse.flow.Flow;
import io.serverlessworkflow.api.types.Workflow;
import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;

@ApplicationScoped
public class UpdateInventoryWorkflow extends Flow {

    @Override
    public Workflow descriptor() {
        return FuncWorkflowBuilder.workflow("updateInventory")
                .tasks(
                        call("updateStock", http("/inventory/{sku}")
                                .method("PUT")
                                .body(".stockUpdate")))
                .build();
    }
}

CREATE TABLE event_deduplication (
    event_id VARCHAR(255) PRIMARY KEY,
    workflow_instance_id VARCHAR(255),
    processed_at TIMESTAMP NOT NULL
);

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.transaction.Transactional;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

@ApplicationScoped
public class DeduplicatingEventConsumer {

    @Inject
    DataSource dataSource;

    @Transactional
    public boolean tryProcess(String eventId) {
        try (Connection conn = dataSource.getConnection()) {
            // Atomically insert — if the key exists, this is a duplicate
            PreparedStatement ps = conn.prepareStatement(
                "INSERT INTO event_deduplication (event_id, processed_at) VALUES (?, NOW())");
            ps.setString(1, eventId);
            try {
                ps.executeUpdate();
                return true; // First time seeing this event
            } catch (java.sql.SQLIntegrityConstraintViolationException e) {
                return false; // Duplicate event, skip
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

// In your event listener or REST endpoint:
if (deduplicator.tryProcess(event.id())) {
    workflowApplication.start(definition, event.payload());
} else {
    // Log and discard the duplicate
}

package org.acme;

import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;

import java.util.Map;

import jakarta.enterprise.context.ApplicationScoped;

import io.quarkiverse.flow.Flow;
import io.serverlessworkflow.api.types.Workflow;
import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;

@ApplicationScoped
public class OrderWorkflow extends Flow {

    @Override
    public Workflow descriptor() {
        return FuncWorkflowBuilder.workflow("orderWorkflow")
                .tasks(
                        call("queueNotification", http("/outbox/notifications")
                                .method("POST")
                                .body(Map.of("to", "${.customer.email}", "subject", "Order confirmed", "orderId", "${.id}"))))
                .build();
    }
}

import jakarta.enterprise.context.ApplicationScoped;
import io.quarkus.scheduler.Scheduled;

@ApplicationScoped
public class OutboxRelay {

    @Scheduled(every = "5s")
    void deliverOutboxMessages() {
        // 1. Read unprocessed messages from the outbox table
        // 2. Attempt delivery (email, HTTP call, Kafka, etc.)
        // 3. Mark as processed on success
        // 4. Implement retry with backoff for failures
    }
}

package org.acme;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;

@ApplicationScoped
public class OutboxExecutionListener implements WorkflowExecutionListener {

    @Inject
    OutboxWriter outboxWriter;

    @Override
    public void onTaskCompleted(TaskCompletedEvent ev) {
        // After certain tasks complete, write a side-effect intent to the outbox
        String taskName = ev.taskContext().taskName();
        if ("queueNotification".equals(taskName)) {
            outboxWriter.write(
                ev.workflowContext().instanceId(),
                "send-email",
                ev.workflowContext().currentData()
            );
        }
    }
}