From 736683da24a782f4c426543a3221047724e7c0ed Mon Sep 17 00:00:00 2001
From: Stefan <91stefan@gmail.com>
Date: Mon, 5 Oct 2026 15:23:35 +0200
Subject: [PATCH 1/3] Add persistent workflow history console and simplify
fulfillment flow
---
order-fulfillment-workflow/.gitignore | 1 +
order-fulfillment-workflow/README.md | 313 ++-
.../docker-compose.yaml | 7 +-
order-fulfillment-workflow/pom.xml | 32 +-
.../InitiatingPaymentForCustomerStarted.java | 2 +-
.../config/WorkflowHistoryConfig.java | 52 +
.../controller/WorkflowHistoryController.java | 63 +
.../JdbcWorkflowHistoryRepository.java | 301 +++
.../history/WorkflowHistoryUpdates.java | 74 +
.../service/FailureRecorder.java | 29 -
.../service/ShippingService.java | 3 +-
.../simulator/AutoPaymentRobot.java | 10 +-
.../workflow/OrderFulfillmentWorkflow.java | 184 +-
.../src/main/resources/application.yaml | 9 +-
.../src/main/resources/static/index.html | 18 +-
.../resources/static/order-fulfillment.bpmn | 235 ++
.../resources/static/vendor/bpmn-js/LICENSE | 23 +
.../resources/static/vendor/bpmn-js/README.md | 12 +
.../static/vendor/bpmn-js/bpmn-js.css | 151 ++
.../bpmn-navigated-viewer.production.min.js | 22 +
.../static/vendor/bpmn-js/diagram-js.css | 1325 +++++++++++
.../main/resources/static/workflow-bpmn.mjs | 160 ++
.../main/resources/static/workflow-trace.mjs | 165 ++
.../src/main/resources/static/workflows.css | 2090 +++++++++++++++++
.../src/main/resources/static/workflows.html | 312 +++
.../src/main/resources/static/workflows.js | 924 ++++++++
.../AbstractOrderFulfillmentTest.java | 106 +
.../JdbcWorkflowHistoryRepositoryTest.java | 188 ++
.../OrderFulfillmentApplicationTest.java | 39 +
.../orderfulfillment/OrderFulfillmentIT.java | 33 +-
.../OrderFulfillmentWorkflowTest.java | 175 ++
.../src/test/js/workflow-trace.test.mjs | 239 ++
32 files changed, 7083 insertions(+), 214 deletions(-)
create mode 100644 order-fulfillment-workflow/.gitignore
create mode 100644 order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/config/WorkflowHistoryConfig.java
create mode 100644 order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/controller/WorkflowHistoryController.java
create mode 100644 order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/history/JdbcWorkflowHistoryRepository.java
create mode 100644 order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/history/WorkflowHistoryUpdates.java
delete mode 100644 order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/service/FailureRecorder.java
create mode 100644 order-fulfillment-workflow/src/main/resources/static/order-fulfillment.bpmn
create mode 100644 order-fulfillment-workflow/src/main/resources/static/vendor/bpmn-js/LICENSE
create mode 100644 order-fulfillment-workflow/src/main/resources/static/vendor/bpmn-js/README.md
create mode 100644 order-fulfillment-workflow/src/main/resources/static/vendor/bpmn-js/bpmn-js.css
create mode 100644 order-fulfillment-workflow/src/main/resources/static/vendor/bpmn-js/bpmn-navigated-viewer.production.min.js
create mode 100644 order-fulfillment-workflow/src/main/resources/static/vendor/bpmn-js/diagram-js.css
create mode 100644 order-fulfillment-workflow/src/main/resources/static/workflow-bpmn.mjs
create mode 100644 order-fulfillment-workflow/src/main/resources/static/workflow-trace.mjs
create mode 100644 order-fulfillment-workflow/src/main/resources/static/workflows.css
create mode 100644 order-fulfillment-workflow/src/main/resources/static/workflows.html
create mode 100644 order-fulfillment-workflow/src/main/resources/static/workflows.js
create mode 100644 order-fulfillment-workflow/src/test/java/io/axoniq/demo/orderfulfillment/AbstractOrderFulfillmentTest.java
create mode 100644 order-fulfillment-workflow/src/test/java/io/axoniq/demo/orderfulfillment/JdbcWorkflowHistoryRepositoryTest.java
create mode 100644 order-fulfillment-workflow/src/test/java/io/axoniq/demo/orderfulfillment/OrderFulfillmentApplicationTest.java
create mode 100644 order-fulfillment-workflow/src/test/java/io/axoniq/demo/orderfulfillment/OrderFulfillmentWorkflowTest.java
create mode 100644 order-fulfillment-workflow/src/test/js/workflow-trace.test.mjs
diff --git a/order-fulfillment-workflow/.gitignore b/order-fulfillment-workflow/.gitignore
new file mode 100644
index 00000000..d0718296
--- /dev/null
+++ b/order-fulfillment-workflow/.gitignore
@@ -0,0 +1 @@
+.data/
diff --git a/order-fulfillment-workflow/README.md b/order-fulfillment-workflow/README.md
index d7619a84..b43a3188 100644
--- a/order-fulfillment-workflow/README.md
+++ b/order-fulfillment-workflow/README.md
@@ -1,95 +1,298 @@
# Order Fulfillment Workflow
-Sample project showing how to model an order fulfillment process using the
-**Axoniq Workflow Engine**. The workflow is written as plain imperative Java —
-the engine handles event sourcing, crash recovery, and audit trails.
+An order fulfillment demo using **Axoniq Workflows** and Spring Boot 4.
+The workflow reserves stock, initiates payment, waits for confirmation, ships the
+order, and notifies the customer. The live map shows shipments moving between
+cities over server-sent events. A second UI at `/workflows.html` inspects persisted
+workflow executions, step state, and the complete workflow event timeline.
+
+## Dependencies
+
+This sample uses Axoniq Framework artifacts from Maven Central:
+
+- `io.axoniq.framework:axoniq-workflow-dsl` provides the Java DSL and workflow engine.
+- `io.axoniq.framework:axoniq-spring-boot-starter` provides workflow infrastructure.
+ This sample registers its `WorkflowModule` explicitly to configure history storage.
+ The former workflow-specific Spring Boot artifact is no longer used.
+- `io.axoniq.framework:axoniq-workflow-test` provides the workflow test fixture.
+- Axon Framework and Axoniq Framework versions are pinned in [`pom.xml`](pom.xml).
+
+Java 21+, Maven 3.9+, and Docker are required to run the complete demo.
+
+## Workflow API
+
+The workflow uses imports under `io.axoniq.framework.workflow`. Its entry point
+reads as the order process, with step implementation in private methods below it:
```java
-@Workflow(idProperty = "orderId", startOnEvent = "io.axoniq.demo.orderfulfillment.api.OrderPlaced")
+@Workflow(idProperty = "orderId",
+ startOnEventClass = OrderPlaced.class,
+ workflowName = "OrderFulfillmentWorkflow")
public void execute(SimpleWorkflowContext ctx) {
- // Register the wait FIRST, so the workflow is subscribed before any payment-confirmation
- // event has a chance to land.
- var paymentConfirmation = ctx.waitForEvent("awaitPayment", PaymentConfirmed.class,
- associate(payloadProperty("orderId"), equalsTo(orderId)), Duration.ofMinutes(15));
-
- var reserved = ctx.awaitExecute("reserveStock", payload, Boolean.class, inventory::reserveStock);
- if (!reserved) {
- paymentConfirmation.cancel("Stock unavailable");
- ctx.fail(new RuntimeException("Stock unavailable"));
+ var paymentConfirmation = registerPaymentWait(ctx);
+
+ if (!reserveStock(ctx)) {
+ rejectOutOfStock(ctx, paymentConfirmation);
return;
}
- ctx.awaitExecute("initiatePayment", payload, payment::initiatePayment,
- Duration.ofSeconds(30),
- baseName("InitiatingPaymentForCustomer").namespace("io.axoniq.demo.orderfulfillment.api"));
-
- paymentConfirmation.await();
- // The Completed event of `shipOrder` (`ShipOrderCompleted`, in the api namespace) is what the
- // projection subscribes to — no eventGateway.publish anywhere inside the workflow.
- var shipResult = ctx.awaitExecute("shipOrder", payload, shipping::shipOrder,
- Duration.ofSeconds(30), namespace("io.axoniq.demo.orderfulfillment.api"));
- ctx.awaitExecute("notifyCustomer", Boolean.class, () -> { notifications.sendConfirmation(email); return true; });
+ awaitPayment(ctx, paymentConfirmation);
+ var trackingNumber = shipOrder(ctx);
+ notifyCustomer(ctx, trackingNumber);
}
```
-## Prerequisites
+[`OrderFulfillmentWorkflow`](src/main/java/io/axoniq/demo/orderfulfillment/workflow/OrderFulfillmentWorkflow.java)
+keeps payload mapping, timeouts, and failure handling in those methods. Payment
+and shipping share a `serviceStep` helper for their timeout and event namespace.
+Their event names and payloads remain stable for the payment robot, shipment
+listeners, and UI. All step IDs remain stable for history and replay.
+
+The payment wait is registered without blocking, before any action that could
+trigger a response. The `awaitPayment` helper initiates payment with nonblocking
+`execute`, joins both results with `allMatch(WorkflowStepResult::success, ...)`,
+and returns only after both succeed. Either can finish first. A failure or timeout
+ends the join, cancels any unfinished companion step, and records the failure.
+Its association helpers come from
+`runtime.association.Associations.associate` and
+`dsl.api.EventAssociationsUtils.equalsTo` / `payloadProperty` within the workflow
+package.
+
+`awaitExecute` with an input payload returns the action's result map. The
+three-argument `awaitExecute(name, resultType, supplier)` convenience is used for
+customer notification.
+
+The engine's `InitiatingPaymentForCustomerStarted` and `ShipOrderCompleted`
+events drive the projection, SSE stream, and simulators. Payment's Started event
+includes the scenario, so the robot can skip forced timeouts without depending on
+projection timing. Shipping generates its tracking number inside the execute
+action and returns it in the recorded result; replay and customer notification
+therefore use the same number.
-- Java 21+
-- Maven 3.9+
-- Docker (for Axon Server)
-- The Axoniq Workflow Engine (`io.axoniq.framework.workflow:*:0.1.0`), available on Maven Central
+Failures use `awaitPublish` to publish `OrderFailed` as a durable workflow step,
+then `ctx.fail(...)` to terminate the workflow. A completed publish is not
+repeated during replay.
-## Running the application
+## Run
+
+From this directory:
```bash
docker compose up -d
mvn spring-boot:run
```
-The application starts on port **9090**.
+Open [the demo UI](http://localhost:9090). Axon Server is available at
+[localhost:8024](http://localhost:8024).
+Compose uses a dedicated project name to keep its data
+separate from older versions of this demo. It persists Axon Server's data, events,
+and replication log so container recreation retains all three.
+Open the [workflow console](http://localhost:9090/workflows.html) directly or follow
+the link in the map header. Its waterfall shows actual step intervals, overlapping
+event waits, and failures. Select a span for its inputs, output, and lifecycle
+events; open the event journal for the full payload and metadata.
-### REST endpoints
+For a live walkthrough, choose **Interactive payment** and **Run workflow**. Watch
+the waiting span grow, then click **Confirm payment** to see shipping and
+notification complete. The wait has a 45-second deadline. **Automatic payment**,
+**Payment timeout**, and **Out of stock** show the other paths.
-| Method | Path | Description |
-|--------|-------------------------------|---------------------------------------------|
-| POST | `/orders?customerId=&email=&amount=` | Place a new order. Returns the order id. |
-| POST | `/orders/{orderId}/payment` | Confirm payment and resume the workflow. |
-| GET | `/orders/{orderId}` | Read the order's projected status. |
+Switch between **Waterfall**, **BPMN**, and **Event journal**. Use **+ / − / Fit**
+to zoom the waterfall or BPMN diagram. **Replay trace** reconstructs state one
+recorded event at a time, with enough playback time to see even short actions.
+The slider and **Previous / Next event** controls move through the same event
+cursor. Every view, including the inspector, shows only the state known at that
+cursor: future steps are pending, results appear upon completion, and rewinding
+removes them. Source timestamps still determine actual durations. Replay executes
+no workflow actions. The
+**Live** button pauses or resumes incoming updates. At narrower widths, the step
+inspector opens as a drawer; close it with **×** or **Escape**.
-### Quick demo
+The BPMN view uses the locally bundled [bpmn-js viewer](https://bpmn.io/toolkit/bpmn-js/walkthrough/)
+and [`order-fulfillment.bpmn`](src/main/resources/static/order-fulfillment.bpmn).
+This illustrative BPMN 2.0 model is authored alongside the Java workflow; it is
+not generated by the framework or used to execute the workflow. It includes the
+parallel payment wait, stock decision, deadline, and failure paths. State colors
+come from the same history cursor as the waterfall. Unvisited branches become
+**Not taken** once the execution finishes. The XML can be downloaded from the view.
-```bash
-# 1. place an order
-ORDER=$(curl -s -X POST 'http://localhost:9090/orders?customerId=alice&email=alice@example.com&amount=99.95')
+The delivery map uses standard OpenStreetMap tiles without an API key and keeps
+visible attribution and normal browser caching. Only the basemap is tinted for
+the dark theme; route and truck colors are unchanged. See the
+[tile usage policy](https://operations.osmfoundation.org/policies/tiles/) for hosting details.
+
+Use a fresh demo event store when switching from the old `0.1.0` sample. This
+sample changes the recorded workflow steps and does not migrate existing workflow
+histories. The map's order projection and its processor tokens remain in memory;
+the truck and payment simulators are demo components. Workflow history and its
+processor token are stored in embedded, file-backed H2.
+
+## Extending the framework history component
+
+Start with [`WorkflowHistoryConfig`](src/main/java/io/axoniq/demo/orderfulfillment/config/WorkflowHistoryConfig.java).
+The sample keeps the framework's standard `WorkflowHistoryProjector` and configures
+its storage and processor through two module extension points:
+
+```java
+WorkflowModule.defaults("OrderFulfillment", SimpleWorkflowContext.class)
+ .definition(definitions -> definitions.autodetected(configuration -> workflow))
+ .withHistory(configuration -> {
+ history.initialize(configuration.getComponent(UnitOfWorkFactory.class));
+ return history;
+ })
+ .historyProcessorConfiguration(processor -> processor
+ .tokenStore(historyTokens)
+ .initialSegmentCount(1)
+ .batchSize(1)
+ .withInterceptor(history));
+```
+
+`axon.workflow.enabled=false` disables automatic workflow module registration;
+Spring registers the explicit module bean instead. It still autodetects the
+workflow definition's `@Workflow` method.
+
+[`JdbcWorkflowHistoryRepository`](src/main/java/io/axoniq/demo/orderfulfillment/history/JdbcWorkflowHistoryRepository.java)
+implements `MutableWorkflowHistoryRepository`. Its processor interceptor provides
+the current event to `save(WorkflowHistory)`, because that repository method only
+receives the resulting state. It stores the event and a UI document in one H2
+transaction. It uses no business `@EventHandler` methods and does not subclass the
+projector.
-# 2. confirm payment
-curl -X POST "http://localhost:9090/orders/$ORDER/payment"
+The event journal has two purposes: it supplies the inspection timeline, and it
+restores the framework's actual state for further projection after a restart. The
+projector requires `EventSourcedWorkflowState`; it cannot continue
+from an arbitrary UI DTO. The repository reconstructs that state by replaying its
+stored events through a fresh standard projector and a temporary in-memory repository.
+This applies recorded state changes only; it does not execute workflow actions.
-# 3. observe the projected status
+The durable data consists of:
+
+| H2 table | Contents |
+|---|---|
+| `workflow_history` | Latest workflow status, payload, version, and step state for the UI. |
+| `workflow_history_event` | Workflow event IDs, type/version, time, payload, and metadata, in recorded order. |
+| `TokenEntry` | The history processor's durable progress and segment claim. |
+
+The demo processes history one event per batch. Event and state writes commit
+before the processor advances its token. They are separate transactions, so an
+interrupted batch may be delivered again. Unique
+event IDs and duplicate checks make that redelivery a no-op. A failed history
+write rolls back both the event and UI state. The original workflow events and
+execution recovery remain in Axon Server; H2 is the inspection read model.
+
+### Live history updates
+
+`WorkflowHistoryConfig` supplies `updates::publish` to the repository. An
+`afterCommit` callback sends the workflow ID to
+[`WorkflowHistoryUpdates`](src/main/java/io/axoniq/demo/orderfulfillment/history/WorkflowHistoryUpdates.java).
+This component coalesces IDs and broadcasts SSE invalidations at
+`/api/workflows/stream`. It observes committed repository writes, using no additional
+business event handlers. Rolled-back writes and duplicate deliveries send nothing.
+Network sends run off the projection thread.
+
+The browser reloads the persisted state and incrementally reads new journal events
+after an invalidation. On reconnect it catches up from the same journal; SSE is
+only a notification channel, not a second history store. Heartbeats maintain the
+connection, with a 30-second reconciliation fetch as a fallback. Running spans
+animate against the server clock between updates; completed spans use their
+recorded end time. Events without a recorded start appear as diamonds, not invented
+durations. Waiting-time totals count overlapping intervals once.
+
+**Extension scope:** the mutable history repository and projector are marked `@Internal`
+in the released framework. This is a version-specific extension example, not a
+framework-provided JDBC implementation. The interceptor relies on the synchronous
+history projector and checks that assumption. Reconstructing state reads all recorded
+events for one workflow on each update; this is suitable for short demo workflows.
+The framework query API uses its query matcher over reconstructed states, while
+the UI reads persisted documents directly with SQL filtering and pagination.
+
+## Persistence and restart demo
+
+The default database is `.data/workflow-history.mv.db`, relative to the process's
+working directory. Start the sample from this directory to keep its location
+consistent. Override `spring.datasource.url` to select another file. No separate
+database container is required.
+
+1. Run a few scenarios at `/workflows.html`, including a failure.
+2. Stop the application, keeping Axon Server and `.data` intact.
+3. Start the application from the same directory and revisit `/workflows.html`.
+4. The execution list and timelines are still available; the history processor
+ resumes from its stored token. New events continue updating the same history.
+
+The timeline shows workflow lifecycle and step events. The independent truck
+location feed remains on the live map; workflow completion precedes truck delivery.
+Keep Axon Server's data and the H2 file together. When resetting this demo to an
+empty event store, also remove `.data` while the application is stopped.
+
+## REST endpoints
+
+| Method | Path | Description |
+|---|---|---|
+| POST | `/orders?customerId=&email=&amount=&scenario=` | Place an order; returns its ID. |
+| POST | `/orders/{orderId}/payment` | Manually confirm payment. |
+| GET | `/orders/{orderId}` | Read the projected order status. |
+| GET | `/orders/stream` | Subscribe to the initial snapshot and live SSE updates. |
+| GET | `/api/workflows?search=&status=&limit=50&offset=0` | Page through persisted executions; search matches workflow ID. |
+| GET | `/api/workflows/stream` | SSE notifications after history commits; reconnect and read the journal to catch up. |
+| GET | `/api/workflows/{workflowId}` | Read persisted workflow and step state. |
+| GET | `/api/workflows/{workflowId}/events?after=0&limit=200` | Read the workflow timeline; `after` is the last returned sequence. |
+| POST | `/simulate/single?scenario=` | Place a random order. |
+| POST | `/simulate/burst?count=10&scenario=` | Place a burst of random orders. |
+| POST | `/simulate/scenario?type=` | Run a selected scenario. |
+
+The default `happy` scenario automatically confirms payment and progresses
+through `PLACED`, `AWAITING_PAYMENT`, `IN_TRANSIT`, and `DELIVERED`.
+`out-of-stock` fails before payment initiation. `payment-timeout` disables the
+payment robot for that order and fails after four seconds; other orders have a
+45-second payment deadline. `manual-payment` also disables the robot and waits for
+an explicit payment confirmation from the console or REST endpoint.
+
+```bash
+# Place an order; the payment robot confirms it automatically.
+ORDER=$(curl -s -X POST 'http://localhost:9090/orders?customerId=alice&email=alice@example.com&amount=99.95')
curl "http://localhost:9090/orders/$ORDER"
+
+# Exercise the two failure paths.
+curl -X POST 'http://localhost:9090/simulate/scenario?type=out-of-stock'
+curl -X POST 'http://localhost:9090/simulate/scenario?type=payment-timeout'
```
-## Integration test
+The workflow completes after shipping and notification. Truck movement and final
+delivery are simulated separately, so `DELIVERED` appears a few seconds later.
+
+## Tests
```bash
+# Workflow fixture and Spring application tests; no Docker required.
+mvn test
+
+# Also run the application against an Axon Server Testcontainer.
mvn verify
-```
-`OrderFulfillmentIT` boots the application against a real Axon Server (started
-via Testcontainers), places an order, confirms payment, and asserts that the
-projection reaches the `SHIPPED` state.
+# Waterfall timing, state replay, and BPMN branch mapping (Node.js 20+).
+node --test src/test/js/workflow-trace.test.mjs
+```
-## What this sample demonstrates
+`OrderFulfillmentWorkflowTest` uses the released `WorkflowTestFixture` and a
+manual clock to verify normal fulfillment, payment arriving before initiation,
+correlation to the correct order, timeout, and stock failure.
+`OrderFulfillmentApplicationTest` checks automatic delivery and the projected
+reasons for both failure scenarios through the HTTP API with an in-memory event
+store, including the history endpoints, scenario timelines, and an interactive
+payment that remains suspended until explicitly confirmed. Tests use isolated
+databases and never touch the demo's `.data` directory. `OrderFulfillmentIT` runs
+the same HTTP scenarios against Axon Server.
-* `@Workflow` with `idProperty` and `startOnEvent`
-* `awaitExecute` with payload and typed return value
-* `awaitExecute` with timeout and event-name customization (`baseName(...)`)
-* `awaitEvent` correlated to the workflow instance via `associate(payloadProperty(...), equalsTo(...))`
-* `ctx.fail(...)` to terminate a workflow with an error
-* A standard `@EventHandler` projection consuming events emitted by the workflow's actions
+`JdbcWorkflowHistoryRepositoryTest` closes and reopens a real H2 file, continues an
+unfinished projection, checks failure/retry/version state, verifies duplicate
+delivery is harmless, and verifies rollback and retry after a failed projection.
+It also verifies that live notifications happen only after successful commits.
## Cleanup
```bash
+# Removes this demo's containers and stored events.
docker compose down -v
+# With the application stopped, also remove its history and processor token.
+rm -rf .data
```
diff --git a/order-fulfillment-workflow/docker-compose.yaml b/order-fulfillment-workflow/docker-compose.yaml
index 9b4ed015..b288ac11 100644
--- a/order-fulfillment-workflow/docker-compose.yaml
+++ b/order-fulfillment-workflow/docker-compose.yaml
@@ -1,7 +1,9 @@
+name: order-fulfillment-workflow-rc1
+
services:
axon-server:
image: docker.axoniq.io/axoniq/axonserver:2026.0.0
- container_name: order-fulfillment-axon-server
+ container_name: order-fulfillment-rc1-axon-server
ports:
- "8024:8024"
- "8124:8124"
@@ -12,9 +14,12 @@ services:
volumes:
- data:/axonserver/data
- events:/axonserver/events
+ - log:/axonserver/log
volumes:
data:
driver: local
events:
driver: local
+ log:
+ driver: local
diff --git a/order-fulfillment-workflow/pom.xml b/order-fulfillment-workflow/pom.xml
index ce4ed52e..6782b4f6 100644
--- a/order-fulfillment-workflow/pom.xml
+++ b/order-fulfillment-workflow/pom.xml
@@ -12,13 +12,11 @@
21
- 21
- 21
+ ${java.version}UTF-8
- 5.3.2
- 5.3.2
- 0.2.0
+ 5.4.0-RC1
+ 5.4.0-RC14.1.12.0.5
@@ -57,11 +55,15 @@
-
+
io.axoniq.framework
- axoniq-workflow-spring-boot
- ${axon-workflow.version}
+ axoniq-workflow-dsl
+ ${axoniq-framework.version}
+
+
+ io.axoniq.framework
+ axoniq-spring-boot-starter
@@ -83,12 +85,21 @@
com.fasterxml.jackson.corejackson-databind
+
+ org.springframework.boot
+ spring-boot-starter-jdbc
+
+
+ com.h2database
+ h2
+ runtime
+ io.axoniq.frameworkaxoniq-workflow-test
- ${axon-workflow.version}
+ ${axoniq-framework.version}test
@@ -140,8 +151,7 @@
maven-compiler-plugin3.16.0
- ${java.version}
- ${java.version}
+ ${java.version}true
diff --git a/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/api/InitiatingPaymentForCustomerStarted.java b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/api/InitiatingPaymentForCustomerStarted.java
index 3c6fc569..c2b77035 100644
--- a/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/api/InitiatingPaymentForCustomerStarted.java
+++ b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/api/InitiatingPaymentForCustomerStarted.java
@@ -6,5 +6,5 @@
* already registered its {@code awaitPayment} wait (the wait is registered at the very top
* of the workflow, before any {@code awaitExecute} step runs).
*/
-public record InitiatingPaymentForCustomerStarted(String orderId) {
+public record InitiatingPaymentForCustomerStarted(String orderId, String scenario) {
}
diff --git a/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/config/WorkflowHistoryConfig.java b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/config/WorkflowHistoryConfig.java
new file mode 100644
index 00000000..e9cb5af3
--- /dev/null
+++ b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/config/WorkflowHistoryConfig.java
@@ -0,0 +1,52 @@
+package io.axoniq.demo.orderfulfillment.config;
+
+import io.axoniq.demo.orderfulfillment.history.JdbcWorkflowHistoryRepository;
+import io.axoniq.demo.orderfulfillment.history.WorkflowHistoryUpdates;
+import io.axoniq.demo.orderfulfillment.workflow.OrderFulfillmentWorkflow;
+import io.axoniq.framework.workflow.configuration.WorkflowModule;
+import io.axoniq.framework.workflow.dsl.simple.SimpleWorkflowContext;
+import org.axonframework.conversion.GeneralConverter;
+import org.axonframework.messaging.core.unitofwork.UnitOfWorkFactory;
+import org.axonframework.messaging.core.unitofwork.transaction.jdbc.JdbcTransactionalExecutorProvider;
+import org.axonframework.messaging.eventhandling.conversion.EventConverter;
+import org.axonframework.messaging.eventhandling.processing.streaming.token.store.jdbc.GenericTokenTableFactory;
+import org.axonframework.messaging.eventhandling.processing.streaming.token.store.jdbc.JdbcTokenStore;
+import org.axonframework.messaging.eventhandling.processing.streaming.token.store.jdbc.JdbcTokenStoreConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+
+/** The two framework extension points demonstrated by this sample. */
+@Configuration(proxyBeanMethods = false)
+public class WorkflowHistoryConfig {
+
+ @Bean
+ JdbcWorkflowHistoryRepository workflowHistoryRepository(DataSource dataSource, EventConverter converter,
+ WorkflowHistoryUpdates updates) {
+ return new JdbcWorkflowHistoryRepository(dataSource, converter, updates::publish);
+ }
+
+ @Bean
+ WorkflowModule orderFulfillmentModule(
+ OrderFulfillmentWorkflow workflow, JdbcWorkflowHistoryRepository history,
+ DataSource dataSource, GeneralConverter converter) {
+ // Only the history processor uses durable tokens. The map projection and demo
+ // simulators retain their existing in-memory token store.
+ var historyTokens = new JdbcTokenStore(new JdbcTransactionalExecutorProvider(dataSource),
+ converter, JdbcTokenStoreConfiguration.DEFAULT);
+ historyTokens.createSchema(GenericTokenTableFactory.INSTANCE);
+
+ return WorkflowModule.defaults("OrderFulfillment", SimpleWorkflowContext.class)
+ .definition(definitions -> definitions.autodetected(configuration -> workflow))
+ .withHistory(configuration -> {
+ history.initialize(configuration.getComponent(UnitOfWorkFactory.class));
+ return history;
+ })
+ .historyProcessorConfiguration(processor -> processor
+ .tokenStore(historyTokens)
+ .initialSegmentCount(1)
+ .batchSize(1)
+ .withInterceptor(history));
+ }
+}
diff --git a/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/controller/WorkflowHistoryController.java b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/controller/WorkflowHistoryController.java
new file mode 100644
index 00000000..f50a9b96
--- /dev/null
+++ b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/controller/WorkflowHistoryController.java
@@ -0,0 +1,63 @@
+package io.axoniq.demo.orderfulfillment.controller;
+
+import io.axoniq.demo.orderfulfillment.history.JdbcWorkflowHistoryRepository;
+import io.axoniq.demo.orderfulfillment.history.WorkflowHistoryUpdates;
+import io.axoniq.framework.workflow.dsl.api.WorkflowStatus;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/workflows")
+public class WorkflowHistoryController {
+
+ private final JdbcWorkflowHistoryRepository history;
+ private final WorkflowHistoryUpdates updates;
+
+ public WorkflowHistoryController(JdbcWorkflowHistoryRepository history, WorkflowHistoryUpdates updates) {
+ this.history = history;
+ this.updates = updates;
+ }
+
+ @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+ public SseEmitter stream() {
+ return updates.subscribe();
+ }
+
+ @GetMapping
+ public Map list(@RequestParam(defaultValue = "") String search,
+ @RequestParam(required = false) WorkflowStatus status,
+ @RequestParam(defaultValue = "50") int limit,
+ @RequestParam(defaultValue = "0") int offset) {
+ if (limit < 1 || limit > 200 || offset < 0) {
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Use limit 1–200 and offset >= 0");
+ }
+ return history.list(search, status == null ? "" : status.name(), limit, offset);
+ }
+
+ @GetMapping("/{workflowId}")
+ public Object detail(@PathVariable String workflowId) {
+ return history.detail(workflowId)
+ .orElseThrow(() -> new ResponseStatusException(HttpStatus.NOT_FOUND, "Workflow not found"));
+ }
+
+ @GetMapping("/{workflowId}/events")
+ public List> events(@PathVariable String workflowId,
+ @RequestParam(defaultValue = "0") long after,
+ @RequestParam(defaultValue = "200") int limit) {
+ if (after < 0 || limit < 1 || limit > 500) {
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Use limit 1–500 and after >= 0");
+ }
+ detail(workflowId);
+ return history.timeline(workflowId, after, limit);
+ }
+}
diff --git a/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/history/JdbcWorkflowHistoryRepository.java b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/history/JdbcWorkflowHistoryRepository.java
new file mode 100644
index 00000000..851a5273
--- /dev/null
+++ b/order-fulfillment-workflow/src/main/java/io/axoniq/demo/orderfulfillment/history/JdbcWorkflowHistoryRepository.java
@@ -0,0 +1,301 @@
+package io.axoniq.demo.orderfulfillment.history;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.core.type.TypeReference;
+import tools.jackson.databind.ObjectMapper;
+import tools.jackson.databind.json.JsonMapper;
+import io.axoniq.framework.workflow.dsl.api.WorkflowState;
+import io.axoniq.framework.workflow.history.api.WorkflowHistory;
+import io.axoniq.framework.workflow.history.inmemory.InMemoryWorkflowHistoryRepository;
+import io.axoniq.framework.workflow.history.inmemory.MutableWorkflowHistoryRepository;
+import io.axoniq.framework.workflow.history.inmemory.WorkflowHistoryProjector;
+import io.axoniq.framework.workflow.query.api.WorkflowStateQuery;
+import io.axoniq.framework.workflow.query.utils.WorkflowStateQueryMatcher;
+import io.axoniq.framework.workflow.runtime.util.MetadataUtils;
+import org.axonframework.messaging.core.MessageHandlerInterceptor;
+import org.axonframework.messaging.core.MessageHandlerInterceptorChain;
+import org.axonframework.messaging.core.MessageStream;
+import org.axonframework.messaging.core.Metadata;
+import org.axonframework.messaging.core.MessageType;
+import org.axonframework.messaging.core.unitofwork.ProcessingContext;
+import org.axonframework.messaging.core.unitofwork.UnitOfWorkFactory;
+import org.axonframework.messaging.eventhandling.EventMessage;
+import org.axonframework.messaging.eventhandling.GenericEventMessage;
+import org.axonframework.messaging.eventhandling.conversion.EventConverter;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import javax.sql.DataSource;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.function.Consumer;
+
+/**
+ * Persists workflow events and the latest workflow state for the history console.
+ *
+ * Each event and its resulting state are stored in a single database transaction.
+ * The console reads the stored state and event journal to display workflow progress.
+ * Stored events also reconstruct the framework's workflow state so history
+ * projection can continue after an application restart.
+ */
+public class JdbcWorkflowHistoryRepository
+ implements MutableWorkflowHistoryRepository, MessageHandlerInterceptor {
+
+ private final JdbcTemplate jdbc;
+ private final TransactionTemplate transactions;
+ private final EventConverter converter;
+ private final Consumer onCommitted;
+ private final ObjectMapper json = JsonMapper.builder().build();
+ private final ThreadLocal projection = new ThreadLocal<>();
+ private UnitOfWorkFactory unitOfWorkFactory;
+
+ public JdbcWorkflowHistoryRepository(DataSource dataSource, EventConverter converter) {
+ this(dataSource, converter, id -> { });
+ }
+
+ public JdbcWorkflowHistoryRepository(DataSource dataSource, EventConverter converter,
+ Consumer onCommitted) {
+ this.jdbc = new JdbcTemplate(dataSource);
+ this.transactions = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
+ this.converter = converter;
+ this.onCommitted = onCommitted;
+ jdbc.execute("""
+ CREATE TABLE IF NOT EXISTS workflow_history (
+ workflow_id VARCHAR(255) PRIMARY KEY,
+ status VARCHAR(40) NOT NULL,
+ updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
+ document CLOB NOT NULL
+ )
+ """);
+ jdbc.execute("""
+ CREATE TABLE IF NOT EXISTS workflow_history_event (
+ sequence BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+ event_id VARCHAR(255) NOT NULL UNIQUE,
+ workflow_id VARCHAR(255) NOT NULL,
+ event_name VARCHAR(1024) NOT NULL,
+ event_version VARCHAR(255) NOT NULL,
+ occurred_at VARCHAR(40) NOT NULL,
+ payload BLOB NOT NULL,
+ metadata CLOB NOT NULL
+ )
+ """);
+ jdbc.execute("CREATE INDEX IF NOT EXISTS history_event_workflow ON workflow_history_event(workflow_id, sequence)");
+ }
+
+ public void initialize(UnitOfWorkFactory unitOfWorkFactory) {
+ this.unitOfWorkFactory = unitOfWorkFactory;
+ }
+
+ @Override
+ public MessageStream interceptOnHandle(EventMessage event, ProcessingContext context,
+ MessageHandlerInterceptorChain chain) {
+ if (!MetadataUtils.hasWorkflowId().test(event.metadata())) {
+ return chain.proceed(event, context);
+ }
+ return transactions.execute(transaction -> {
+ // History can commit before the processor's token. Redelivery must be a no-op.
+ if (Boolean.TRUE.equals(jdbc.queryForObject(
+ "SELECT EXISTS(SELECT 1 FROM workflow_history_event WHERE event_id = ?)",
+ Boolean.class, event.identifier()))) {
+ return MessageStream.empty();
+ }
+ projection.set(new Projection(event, context));
+ try {
+ var result = chain.proceed(event, context);
+ // The standard projector performs findById/evolve/save synchronously.
+ // Fail explicitly if a future framework version changes that contract.
+ if (!result.isCompleted()) {
+ throw new IllegalStateException("History adapter requires a synchronous history projector");
+ }
+ if (result.error().isPresent()) {
+ throw new IllegalStateException("History projection failed", result.error().get());
+ }
+ return result;
+ } finally {
+ projection.remove();
+ }
+ });
+ }
+
+ @Override
+ public void save(WorkflowHistory history) {
+ var active = projection.get();
+ if (active == null) {
+ throw new IllegalStateException("Configure this repository as the history processor interceptor too");
+ }
+ var event = active.event();
+ if (!history.workflowId().equals(MetadataUtils.getWorkflowId(event.metadata()))) {
+ throw new IllegalArgumentException("History and event workflow IDs differ");
+ }
+ jdbc.update("""
+ INSERT INTO workflow_history_event
+ (event_id, workflow_id, event_name, event_version, occurred_at, payload, metadata)
+ VALUES (?, ?, ?, ?, ?, ?, ?)
+ """, event.identifier(), history.workflowId(), event.type().qualifiedName().toString(),
+ event.type().version(), event.timestamp().toString(), event.payloadAs(byte[].class),
+ write(event.metadata()));
+ jdbc.update("""
+ MERGE INTO workflow_history (workflow_id, status, updated_at, document)
+ KEY (workflow_id) VALUES (?, ?, ?, ?)
+ """, history.workflowId(), history.state().workflowStatus().name(),
+ OffsetDateTime.ofInstant(event.timestamp(), ZoneOffset.UTC), write(document(history.state(), event)));
+ TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+ @Override
+ public void afterCommit() {
+ onCommitted.accept(history.workflowId());
+ }
+ });
+ }
+
+ @Override
+ public CompletableFuture> findById(String workflowId) {
+ return withContext(context -> restore(workflowId, context));
+ }
+
+ @Override
+ public CompletableFuture> findAll(WorkflowStateQuery query) {
+ // For this small demo, evaluate the framework's complete query semantics in Java.
+ // The UI uses the indexed SQL list below and does not replay every workflow.
+ return withContext(context -> jdbc.queryForList("SELECT workflow_id FROM workflow_history", String.class)
+ .stream().map(id -> restore(id, context)).flatMap(Optional::stream)
+ .filter(history -> WorkflowStateQueryMatcher.matches(query, history.state())).toList());
+ }
+
+ private Optional restore(String workflowId, ProcessingContext context) {
+ var memory = new InMemoryWorkflowHistoryRepository();
+ var projector = new WorkflowHistoryProjector(memory);
+ jdbc.query("SELECT * FROM workflow_history_event WHERE workflow_id = ? ORDER BY sequence",
+ (org.springframework.jdbc.core.RowCallbackHandler) row -> projector.handle(readEvent(row), context),
+ workflowId);
+ return memory.findById(workflowId).join();
+ }
+
+ private CompletableFuture withContext(Function operation) {
+ var active = projection.get();
+ if (active != null) {
+ return CompletableFuture.completedFuture(operation.apply(active.context()));
+ }
+ if (unitOfWorkFactory == null) {
+ return CompletableFuture.failedFuture(new IllegalStateException("History repository is not initialized"));
+ }
+ return unitOfWorkFactory.create().executeWithResult(
+ context -> CompletableFuture.completedFuture(operation.apply(context)));
+ }
+
+ @Override
+ public void clear() {
+ transactions.executeWithoutResult(transaction -> {
+ jdbc.update("DELETE FROM workflow_history_event");
+ jdbc.update("DELETE FROM workflow_history");
+ });
+ }
+
+ public Map list(String search, String status, int limit, int offset) {
+ var pattern = "%" + search.toLowerCase(java.util.Locale.ROOT)
+ .replace("!", "!!").replace("%", "!%").replace("_", "!_") + "%";
+ var where = " WHERE LOWER(workflow_id) LIKE ? ESCAPE '!' AND (? = '' OR status = ?)";
+ var items = jdbc.query("SELECT document FROM workflow_history" + where
+ + " ORDER BY updated_at DESC, workflow_id LIMIT ? OFFSET ?",
+ (row, index) -> read(row.getString(1)), pattern, status, status, limit, offset);
+ var total = jdbc.queryForObject("SELECT COUNT(*) FROM workflow_history" + where,
+ Long.class, pattern, status, status);
+ return Map.of("items", items, "total", total, "limit", limit, "offset", offset, "storage", "H2 file");
+ }
+
+ public Optional