# Audit Trail System (/learn/events-store/scenarios/audit-trail)



This scenario builds a compliance-ready audit trail that records every user action as an immutable event. Events Store's persistence, sequencing, and replay capabilities make it ideal for audit logging where data integrity and traceability are mandatory.

## Architecture [#architecture]

<Mermaid
  chart="graph LR
  API[&#x22;API Gateway&#x22;]
  AUTH[&#x22;Auth Service&#x22;]
  PAY[&#x22;Payment Service&#x22;]
  ES[(&#x22;Events Store<br/>audit.user-actions&#x22;)]
  COMP[&#x22;Compliance Dashboard<br/>StartFromFirst&#x22;]
  SEC[&#x22;Security Monitor<br/>StartAtTimeDelta&#x22;]
  SIEM[&#x22;SIEM Integration<br/>StartNewOnly&#x22;]

  API -- &#x22;log action&#x22; --> ES
  AUTH -- &#x22;log action&#x22; --> ES
  PAY -- &#x22;log action&#x22; --> ES

  ES -- &#x22;replay all&#x22; --> COMP
  ES -- &#x22;last 24h&#x22; --> SEC
  ES -. &#x22;live only&#x22; .-> SIEM

  class API,AUTH,PAY,COMP,SEC,SIEM client
  class ES store"
/>

*Every service writes to one immutable audit channel; each consumer replays from a different start position.*

### Design Decisions [#design-decisions]

* **Single audit channel** — all services publish to `audit.user-actions` for a unified timeline
* **Immutable events** — once stored, events cannot be modified or deleted
* **Sequence numbers** — provide a tamper-evident, monotonically increasing ordering
* **Multiple consumers** — compliance, security, and SIEM systems each subscribe with different start positions

## Implementation [#implementation]

<Steps>
  <Step>
    ### Define the Audit Event Schema [#define-the-audit-event-schema]

    Every audit event follows a consistent structure for compliance tooling:

    ```json
    {
      "eventType": "user.data.export",
      "actor": "user:U-1001",
      "resource": "orders:ORD-5001",
      "action": "export",
      "outcome": "success",
      "ip": "192.168.1.42",
      "userAgent": "Mozilla/5.0...",
      "timestamp": "2026-03-26T14:30:00Z",
      "metadata": {
        "exportFormat": "csv",
        "recordCount": 150
      }
    }
    ```
  </Step>

  <Step>
    ### Publish Audit Events from Services [#publish-audit-events-from-services]

    Each service publishes audit events as part of its request handling.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="audit_logger.go"
        func logAuditEvent(ctx context.Context, client *kubemq.Client, event AuditEvent) error {
            body, _ := json.Marshal(event)
            result, err := client.SendEventStore(ctx, kubemq.NewEvent().
                SetChannel("audit.user-actions").
                SetMetadata(event.EventType).
                SetBody(body).
                SetTags(map[string]string{
                    "actor":    event.Actor,
                    "resource": event.Resource,
                    "outcome":  event.Outcome,
                }),
            )
            if err != nil {
                return fmt.Errorf("audit log failed: %w", err)
            }
            log.Printf("Audit logged: %s seq=%s", event.EventType, result.EventID)
            return nil
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="audit_logger.py"
        def log_audit_event(client, event: dict) -> None:
            result = client.publish_event_store(
                EventStoreMessage(
                    channel="audit.user-actions",
                    metadata=event["eventType"],
                    body=json.dumps(event).encode("utf-8"),
                    tags={
                        "actor": event["actor"],
                        "resource": event["resource"],
                        "outcome": event["outcome"],
                    },
                )
            )
            print(f"Audit logged: {event['eventType']} ID={result.id}")
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="audit_logger.ts"
        async function logAuditEvent(client: KubeMQClient, event: AuditEvent) {
          const result = await client.sendEventStore(
            createEventStoreMessage({
              channel: 'audit.user-actions',
              metadata: event.eventType,
              body: JSON.stringify(event),
              tags: {
                actor: event.actor,
                resource: event.resource,
                outcome: event.outcome,
              },
            })
          );
          console.log(`Audit logged: ${event.eventType} ID=${result.id}`);
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="AuditLogger.java"
        public void logAuditEvent(PubSubClient client, AuditEvent event) {
            var result = client.sendEventsStoreMessage(
                EventStoreMessage.builder()
                    .channel("audit.user-actions")
                    .metadata(event.getEventType())
                    .body(objectMapper.writeValueAsBytes(event))
                    .tags(Map.of(
                        "actor", event.getActor(),
                        "resource", event.getResource(),
                        "outcome", event.getOutcome()))
                    .build());
            System.out.printf("Audit logged: %s ID=%s%n", event.getEventType(), result.getId());
        }
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="AuditLogger.cs"
        public async Task LogAuditEventAsync(KubeMQClient client, AuditEvent auditEvent)
        {
            var result = await client.SendEventStoreAsync(new EventStoreMessage
            {
                Channel = "audit.user-actions",
                Metadata = auditEvent.EventType,
                Body = JsonSerializer.SerializeToUtf8Bytes(auditEvent),
                Tags = { ["actor"] = auditEvent.Actor, ["resource"] = auditEvent.Resource },
            });
            Console.WriteLine($"Audit logged: {auditEvent.EventType} ID={result.Id}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="AuditLogger.kt"
        suspend fun logAuditEvent(client: KubeMQClient, event: AuditEvent) {
            val result = client.sendEventStore(eventStoreMessage {
                channel = "audit.user-actions"
                metadata = event.eventType
                body = Json.encodeToString(event).toByteArray()
                tags = mapOf("actor" to event.actor, "resource" to event.resource)
            })
            println("Audit logged: ${event.eventType} ID=${result.id}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="audit_logger.cc"
        void log_audit_event(kubemq::Client& client, const AuditEvent& event) {
            kubemq::EventStoreMessage msg;
            msg.set_channel("audit.user-actions");
            msg.set_metadata(event.event_type);
            msg.set_body(event.to_json());
            msg.set_tag("actor", event.actor);
            msg.set_tag("resource", event.resource);
            auto result = client.SendEventStore(msg);
            if (result.ok()) {
                std::cout << "Audit logged: " << event.event_type << std::endl;
            }
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="audit_logger.rs"
        async fn log_audit_event(client: &KubemqClient, event: &AuditEvent) -> kubemq::Result<()> {
            let store_event = EventStoreBuilder::new()
                .channel("audit.user-actions")
                .metadata(&event.event_type)
                .body(serde_json::to_vec(event).unwrap())
                .add_tag("actor", &event.actor)
                .add_tag("resource", &event.resource)
                .add_tag("outcome", &event.outcome)
                .build();

            let result = client.send_event_store(store_event).await?;
            println!("Audit logged: {} id={}", event.event_type, result.id);
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="audit_logger.rb"
        def log_audit_event(client, event)
          msg = KubeMQ::PubSub::EventStoreMessage.new(
            channel: 'audit.user-actions',
            metadata: event[:event_type],
            body: event.to_json,
            tags: {
              'actor' => event[:actor],
              'resource' => event[:resource],
              'outcome' => event[:outcome]
            }
          )
          result = client.send_event_store(msg)
          puts "Audit logged: #{event[:event_type]} sent=#{result.sent}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="audit_logger.exs"
        def log_audit_event(client, event) do
          store_event =
            KubeMQ.EventStore.new(
              channel: "audit.user-actions",
              metadata: event.event_type,
              body: Jason.encode!(event),
              tags: %{
                "actor" => event.actor,
                "resource" => event.resource,
                "outcome" => event.outcome
              }
            )

          case KubeMQ.Client.send_event_store(client, store_event) do
            {:ok, result} -> IO.puts("Audit logged: #{event.event_type} id=#{result.id}")
            {:error, err} -> IO.puts("Audit log failed: #{err.message}")
          end
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Subscribe for Compliance Replay [#subscribe-for-compliance-replay]

    The compliance dashboard replays the full audit history using `StartFromFirst`.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="compliance_dashboard.go"
        sub, err := client.SubscribeToEventsStore(ctx,
            "audit.user-actions",
            "compliance-reader",
            kubemq.StartFromFirst(),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[Compliance] seq=%d actor=%s action=%s\n",
                    event.Sequence, event.Tags["actor"], event.Metadata)
            }),
            kubemq.WithOnError(func(err error) {
                log.Println("Error:", err)
            }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="compliance_dashboard.py"
        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="audit.user-actions",
                group="compliance-reader",
                start_position=EventStoreStartPosition.StartFromFirst,
                on_receive_event_callback=lambda e: print(
                    f"[Compliance] seq={e.sequence} actor={e.tags.get('actor')} "
                    f"action={e.metadata}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="compliance_dashboard.ts"
        client.subscribeToEventsStore({
          channel: 'audit.user-actions',
          group: 'compliance-reader',
          startPosition: EventStoreStartPosition.StartFromFirst,
          onEvent: (msg) =>
            console.log(`[Compliance] seq=${msg.sequence} actor=${msg.tags?.actor} action=${msg.metadata}`),
          onError: (err) => console.error('Error:', err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ComplianceDashboard.java"
        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("audit.user-actions")
            .group("compliance-reader")
            .startPosition(EventStoreStartPosition.StartFromFirst)
            .onReceiveEventCallback(event ->
                System.out.printf("[Compliance] seq=%d actor=%s action=%s%n",
                    event.getSequence(), event.getTags().get("actor"), event.getMetadata()))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ComplianceDashboard.cs"
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "audit.user-actions",
                Group = "compliance-reader",
                StartPosition = EventStoreStartPosition.StartFromFirst,
            }))
        {
            Console.WriteLine($"[Compliance] seq={msg.Sequence} action={msg.Metadata}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ComplianceDashboard.kt"
        client.subscribeToEventsStore {
            channel = "audit.user-actions"
            group = "compliance-reader"
            startPosition = StartPosition.StartFromFirst
        }.collect { msg ->
            println("[Compliance] seq=${msg.sequence} action=${msg.metadata}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="compliance_dashboard.cc"
        client->SubscribeToEventsStore("audit.user-actions", "compliance-reader",
            kubemq::StartPosition::StartFromFirst,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[Compliance] seq=" << msg.sequence()
                          << " action=" << msg.metadata() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="compliance_dashboard.rs"
        let sub = client
            .subscribe_to_events_store(
                "audit.user-actions",
                "compliance-reader",
                EventsStoreSubscription::StartFromFirst,
                |event| {
                    Box::pin(async move {
                        println!(
                            "[Compliance] seq={} actor={} action={}",
                            event.sequence,
                            event.tags.get("actor").map(String::as_str).unwrap_or(""),
                            event.metadata,
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="compliance_dashboard.rb"
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'audit.user-actions',
          group: 'compliance-reader',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
        )

        client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
          puts "Error: #{e.message}"
        }) do |event|
          puts "[Compliance] seq=#{event.sequence} actor=#{event.tags['actor']} action=#{event.metadata}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="compliance_dashboard.exs"
        {:ok, _sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "audit.user-actions",
            group: "compliance-reader",
            start_at: :start_from_first,
            on_event: fn event ->
              IO.puts(
                "[Compliance] seq=#{event.sequence} " <>
                  "actor=#{Map.get(event.tags, "actor")} action=#{event.metadata}"
              )
            end
          )
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="Retention Policy">
    Set `Store.MaxRetention` to match your compliance requirements. For regulated industries, use unlimited retention (`0`) or a value exceeding your legal retention window (e.g., 7 years). Combine with adequate disk capacity.
  </Accordion>

  <Accordion title="Tamper Evidence">
    Events Store sequence numbers provide a tamper-evident ordering. If a sequence gap is detected during replay, it indicates data loss or corruption. Implement gap detection in your compliance reader.
  </Accordion>

  <Accordion title="Multi-Region Considerations">
    For multi-region deployments, publish to a local KubeMQ instance and use a separate process to replicate audit events across regions. This ensures audit logging doesn't add cross-region latency to the request path.
  </Accordion>

  <Accordion title="Performance Impact">
    Audit logging adds a single Events Store publish per request. This typically adds 1-3ms of latency. For ultra-low-latency paths, consider publishing audit events asynchronously (fire-and-forget to a local buffer that flushes to Events Store).
  </Accordion>
</Accordions>

## Related [#related]

* [Configure Retention](/learn/events-store/how-to/configure-retention) for compliance retention windows
* [Event Sourcing](/learn/events-store/tutorials/event-sourcing) for state reconstruction from audit logs
* [Events Store Reference](/learn/events-store/reference) for message structure details