# Event-Driven State Machine (/learn/events-store/scenarios/event-driven-state-machine)



This scenario implements an order processing state machine where each state transition is recorded as a persistent event. The full order lifecycle — from creation through delivery — is captured as an immutable event stream that can be replayed to rebuild state at any point.

## Architecture [#architecture]

Each service publishes a transition to a per-order Events Store channel. The append-only stream is the source of truth; any consumer can replay it from the first sequence to rebuild the current state.

<Mermaid
  chart="graph LR
  PAY[&#x22;Payment service&#x22;]
  WH[&#x22;Warehouse worker&#x22;]
  SHIP[&#x22;Shipping service&#x22;]
  CH{{&#x22;Events Store channel<br/>order-state.ORD-2001&#x22;}}
  STORE[&#x22;Persistent log<br/>seq 1..N (append-only)&#x22;]
  RB[&#x22;State rebuilder<br/>StartFromFirst&#x22;]

  PAY -- &#x22;order.paid&#x22; --> CH
  WH -- &#x22;order.picking&#x22; --> CH
  SHIP -- &#x22;order.shipped&#x22; --> CH
  CH --> STORE
  STORE -. &#x22;replay 1..N&#x22; .-> RB

  class PAY,WH,SHIP,RB client
  class CH events
  class STORE store"
/>

*Transitions fan into one per-order channel; the persistent log is replayed to recover the latest state.*

## State Machine [#state-machine]

<Mermaid
  chart="stateDiagram-v2
    [*] --> Created: order.created
    Created --> Paid: order.paid
    Created --> Cancelled: order.cancelled
    Paid --> Picking: order.picking
    Picking --> Shipped: order.shipped
    Shipped --> Delivered: order.delivered
    Shipped --> ReturnRequested: order.return_requested
    ReturnRequested --> Returned: order.returned

    Created: New order placed
    Paid: Payment confirmed
    Picking: Items being picked
    Shipped: In transit
    Delivered: Complete
    Cancelled: Order cancelled
    ReturnRequested: Return initiated
    Returned: Return completed"
/>

*The legal transition graph — the rebuilder applies stored events in sequence to walk these edges.*

### Valid Transitions [#valid-transitions]

| From            | To              | Event                    |
| --------------- | --------------- | ------------------------ |
| —               | Created         | `order.created`          |
| Created         | Paid            | `order.paid`             |
| Created         | Cancelled       | `order.cancelled`        |
| Paid            | Picking         | `order.picking`          |
| Picking         | Shipped         | `order.shipped`          |
| Shipped         | Delivered       | `order.delivered`        |
| Shipped         | ReturnRequested | `order.return_requested` |
| ReturnRequested | Returned        | `order.returned`         |

## Implementation [#implementation]

<Steps>
  <Step>
    ### Publish State Transition Events [#publish-state-transition-events]

    Each state change is published as a persistent event to a per-order channel.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="order_state_machine.go"
        package main

        import (
            "context"
            "fmt"
            "log"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        type StateTransition struct {
            OrderID   string `json:"orderId"`
            FromState string `json:"fromState"`
            ToState   string `json:"toState"`
            Event     string `json:"event"`
            Actor     string `json:"actor"`
        }

        func transitionOrder(ctx context.Context, client *kubemq.Client, t StateTransition) error {
            channel := fmt.Sprintf("order-state.%s", t.OrderID)
            body := fmt.Sprintf(`{"orderId":"%s","fromState":"%s","toState":"%s","event":"%s","actor":"%s"}`,
                t.OrderID, t.FromState, t.ToState, t.Event, t.Actor)

            result, err := client.SendEventStore(ctx, kubemq.NewEvent().
                SetChannel(channel).
                SetMetadata(t.Event).
                SetBody([]byte(body)),
            )
            if err != nil {
                return fmt.Errorf("transition failed: %w", err)
            }
            log.Printf("Transition: %s -> %s (seq: %s)", t.FromState, t.ToState, result.EventID)
            return nil
        }

        func main() {
            ctx := context.Background()
            client, _ := kubemq.NewClient(ctx, kubemq.WithAddress("localhost", 50000))
            defer client.Close()

            transitions := []StateTransition{
                {"ORD-2001", "", "created", "order.created", "customer:C-100"},
                {"ORD-2001", "created", "paid", "order.paid", "payment-service"},
                {"ORD-2001", "paid", "picking", "order.picking", "warehouse-worker:W-5"},
                {"ORD-2001", "picking", "shipped", "order.shipped", "shipping-service"},
                {"ORD-2001", "shipped", "delivered", "order.delivered", "carrier:fedex"},
            }

            for _, t := range transitions {
                transitionOrder(ctx, client, t)
            }
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="order_state_machine.py"
        import json
        from kubemq import PubSubClient, EventStoreMessage

        def transition_order(client, order_id, from_state, to_state, event, actor):
            body = json.dumps({
                "orderId": order_id, "fromState": from_state,
                "toState": to_state, "event": event, "actor": actor,
            })
            result = client.publish_event_store(
                EventStoreMessage(
                    channel=f"order-state.{order_id}",
                    metadata=event,
                    body=body.encode("utf-8"),
                )
            )
            print(f"Transition: {from_state} -> {to_state} (ID: {result.id})")

        with PubSubClient(address="localhost:50000") as client:
            transitions = [
                ("ORD-2001", "", "created", "order.created", "customer:C-100"),
                ("ORD-2001", "created", "paid", "order.paid", "payment-service"),
                ("ORD-2001", "paid", "picking", "order.picking", "warehouse:W-5"),
                ("ORD-2001", "picking", "shipped", "order.shipped", "shipping-service"),
                ("ORD-2001", "shipped", "delivered", "order.delivered", "carrier:fedex"),
            ]
            for oid, fs, ts, evt, actor in transitions:
                transition_order(client, oid, fs, ts, evt, actor)
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="order_state_machine.ts"
        import { KubeMQClient, createEventStoreMessage } from 'kubemq-js';

        const client = await KubeMQClient.create({ address: 'localhost:50000' });

        interface StateTransition {
          orderId: string; fromState: string; toState: string; event: string; actor: string;
        }

        async function transitionOrder(t: StateTransition) {
          const result = await client.sendEventStore(
            createEventStoreMessage({
              channel: `order-state.${t.orderId}`,
              metadata: t.event,
              body: JSON.stringify(t),
            })
          );
          console.log(`Transition: ${t.fromState} -> ${t.toState} (ID: ${result.id})`);
        }

        const transitions: StateTransition[] = [
          { orderId: 'ORD-2001', fromState: '', toState: 'created', event: 'order.created', actor: 'customer:C-100' },
          { orderId: 'ORD-2001', fromState: 'created', toState: 'paid', event: 'order.paid', actor: 'payment-service' },
          { orderId: 'ORD-2001', fromState: 'paid', toState: 'picking', event: 'order.picking', actor: 'warehouse:W-5' },
          { orderId: 'ORD-2001', fromState: 'picking', toState: 'shipped', event: 'order.shipped', actor: 'shipping-service' },
          { orderId: 'ORD-2001', fromState: 'shipped', toState: 'delivered', event: 'order.delivered', actor: 'carrier:fedex' },
        ];

        for (const t of transitions) await transitionOrder(t);
        ```
      </Tab>

      <Tab value="Java">
        ```java title="OrderStateMachine.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("state-machine")
            .build();

        String[][] transitions = {
            {"ORD-2001", "", "created", "order.created", "customer:C-100"},
            {"ORD-2001", "created", "paid", "order.paid", "payment-service"},
            {"ORD-2001", "paid", "picking", "order.picking", "warehouse:W-5"},
            {"ORD-2001", "picking", "shipped", "order.shipped", "shipping-service"},
            {"ORD-2001", "shipped", "delivered", "order.delivered", "carrier:fedex"},
        };

        for (String[] t : transitions) {
            String body = String.format(
                "{\"orderId\":\"%s\",\"fromState\":\"%s\",\"toState\":\"%s\",\"event\":\"%s\"}",
                t[0], t[1], t[2], t[3]);
            client.sendEventsStoreMessage(EventStoreMessage.builder()
                .channel("order-state." + t[0])
                .metadata(t[3])
                .body(body.getBytes())
                .build());
            System.out.printf("Transition: %s -> %s%n", t[1], t[2]);
        }
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="OrderStateMachine.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        var transitions = new[] {
            ("", "created", "order.created"),
            ("created", "paid", "order.paid"),
            ("paid", "picking", "order.picking"),
            ("picking", "shipped", "order.shipped"),
            ("shipped", "delivered", "order.delivered"),
        };

        foreach (var (from, to, evt) in transitions)
        {
            var body = $"{{\"orderId\":\"ORD-2001\",\"fromState\":\"{from}\",\"toState\":\"{to}\"}}";
            await client.SendEventStoreAsync(new EventStoreMessage
            {
                Channel = "order-state.ORD-2001",
                Metadata = evt,
                Body = Encoding.UTF8.GetBytes(body),
            });
            Console.WriteLine($"Transition: {from} -> {to}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="OrderStateMachine.kt"
        val client = KubeMQClient.pubSub {
            address = "localhost:50000"
            clientId = "state-machine"
        }

        data class Transition(val from: String, val to: String, val event: String)

        val transitions = listOf(
            Transition("", "created", "order.created"),
            Transition("created", "paid", "order.paid"),
            Transition("paid", "picking", "order.picking"),
            Transition("picking", "shipped", "order.shipped"),
            Transition("shipped", "delivered", "order.delivered"),
        )

        client.use {
            for (t in transitions) {
                client.sendEventStore(eventStoreMessage {
                    channel = "order-state.ORD-2001"
                    metadata = t.event
                    body = """{"fromState":"${t.from}","toState":"${t.to}"}""".toByteArray()
                })
                println("Transition: ${t.from} -> ${t.to}")
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="order_state_machine.cc"
        kubemq::ClientOptions options;
        options.set_address("localhost", 50000);
        auto client = kubemq::Client::Create(options).value();

        struct Transition { std::string from, to, event; };
        std::vector<Transition> transitions = {
            {"", "created", "order.created"},
            {"created", "paid", "order.paid"},
            {"paid", "picking", "order.picking"},
            {"picking", "shipped", "order.shipped"},
            {"shipped", "delivered", "order.delivered"},
        };

        for (const auto& t : transitions) {
            kubemq::EventStoreMessage msg;
            msg.set_channel("order-state.ORD-2001");
            msg.set_metadata(t.event);
            msg.set_body("{\"fromState\":\"" + t.from + "\",\"toState\":\"" + t.to + "\"}");
            client->SendEventStore(msg);
            std::cout << "Transition: " << t.from << " -> " << t.to << std::endl;
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="order_state_machine.rs"
        use kubemq::prelude::*;
        use kubemq::EventStoreBuilder;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            let order_id = "ORD-2001";
            let channel = format!("order-state.{order_id}");

            let transitions = [
                ("", "created", "order.created"),
                ("created", "paid", "order.paid"),
                ("paid", "picking", "order.picking"),
                ("picking", "shipped", "order.shipped"),
                ("shipped", "delivered", "order.delivered"),
            ];

            for (from, to, event) in transitions {
                let body = format!(
                    r#"{{"orderId":"{order_id}","fromState":"{from}","toState":"{to}"}}"#
                );
                let msg = EventStoreBuilder::new()
                    .channel(&channel)
                    .metadata(event)
                    .body(body.into_bytes())
                    .build();

                let result = client.send_event_store(msg).await?;
                println!("Transition: {from} -> {to} (id: {}, sent: {})", result.id, result.sent);
            }

            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="order_state_machine.rb"
        require 'kubemq'

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'state-machine')
        order_id = 'ORD-2001'

        transitions = [
          ['', 'created', 'order.created'],
          ['created', 'paid', 'order.paid'],
          ['paid', 'picking', 'order.picking'],
          ['picking', 'shipped', 'order.shipped'],
          ['shipped', 'delivered', 'order.delivered']
        ]

        transitions.each do |from, to, event|
          body = { orderId: order_id, fromState: from, toState: to }.to_json
          msg = KubeMQ::PubSub::EventStoreMessage.new(
            channel: "order-state.#{order_id}",
            metadata: event,
            body: body
          )
          result = client.send_event_store(msg)
          puts "Transition: #{from} -> #{to} (id: #{result.id}, sent: #{result.sent})"
        end

        client.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="order_state_machine.exs"
        {:ok, client} =
          KubeMQ.Client.start_link(address: "localhost:50000", client_id: "state-machine")

        order_id = "ORD-2001"

        transitions = [
          {"", "created", "order.created"},
          {"created", "paid", "order.paid"},
          {"paid", "picking", "order.picking"},
          {"picking", "shipped", "order.shipped"},
          {"shipped", "delivered", "order.delivered"}
        ]

        for {from, to, event} <- transitions do
          body = Jason.encode!(%{orderId: order_id, fromState: from, toState: to})

          event_msg =
            KubeMQ.EventStore.new(
              channel: "order-state.#{order_id}",
              metadata: event,
              body: body
            )

          {:ok, result} = KubeMQ.Client.send_event_store(client, event_msg)
          IO.puts("Transition: #{from} -> #{to} (id: #{result.id}, sent: #{result.sent})")
        end

        KubeMQ.Client.close(client)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Rebuild Current State [#rebuild-current-state]

    Subscribe with `StartFromFirst` to replay the transition history and determine the current state.

    ```text
    seq=1 order.created  -> state=created
    seq=2 order.paid     -> state=paid
    seq=3 order.picking  -> state=picking
    seq=4 order.shipped  -> state=shipped
    seq=5 order.delivered -> state=delivered

    Current state: delivered
    ```
  </Step>
</Steps>

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="Transition Validation">
    Validate state transitions **before** publishing. Reject invalid transitions (e.g., `created -> shipped` skipping `paid`). The event store is append-only — you cannot undo an invalid event. If an invalid transition is published, append a compensating event (e.g., `order.transition_reverted`).
  </Accordion>

  <Accordion title="Concurrency">
    Use a single publisher per order or implement optimistic concurrency by checking the expected sequence number before publishing. If the sequence has advanced, another process made a transition — reload state and retry.
  </Accordion>

  <Accordion title="Snapshotting for Long-Lived Orders">
    For orders with many state changes, periodically save a snapshot of the current state alongside the last processed sequence number. On startup, load the snapshot and replay only subsequent events.
  </Accordion>

  <Accordion title="Channel Naming Strategy">
    Use `order-state.{orderId}` for per-order channels. This provides independent replay per order and avoids cross-order interference. For aggregate queries (e.g., "all orders in shipped state"), maintain a separate read model updated by a consumer group.
  </Accordion>
</Accordions>

## Related [#related]

* [Event Sourcing](/learn/events-store/tutorials/event-sourcing) for the foundational pattern
* [Replay Events](/learn/events-store/tutorials/replay-events) for state reconstruction techniques
* [Cross-Service Sync](/learn/events-store/scenarios/cross-service-sync) for multi-service state coordination