System Events

Verified by tests

SystemEventEmitterTests, SystemEventTransportFilterTests, SystemEventOptionsTests, AuditingEventStoreDecoratorTests, CommandAuditPipelineBehaviorTests, SystemEventServiceCollectionExtensionsTests, SystemEventSelfAuditPreventionTests, EventAuditedTests, SecuritySystemEventTests — library CI run #31657041675 (2026-08-13)

System events are internal events emitted by Whizbang for observability, auditing, and diagnostics. Unlike domain events which represent business facts, system events capture infrastructure operations, security decisions, and audit trails.

Core Concept

graph TB
    DI["Domain Infrastructure (Events, Commands, Perspectives)"]
    EM["System Event Emitter"]
    E1["EventAudited (domain event stored)"]
    E2["CommandAudited (command processed)"]
    E3["ScopeContextEstablished (security context set)"]
    E4["AccessGranted/AccessDenied (authorization)"]
    E5["PermissionChanged (role/permission changes)"]
    S1["Stored in dedicated $wb-system stream"]
    S2["Consumed by perspectives (same as domain events)"]
    S3["LocalOnly by default (no network traffic)"]

    DI --> EM
    EM --> E1
    EM --> E2
    EM --> E3
    EM --> E4
    EM --> E5
    E1 --> S1
    E2 --> S1
    E3 --> S1
    E4 --> S1
    E5 --> S1
    S1 --> S2 --> S3

    style DI fill:#d4edda,stroke:#28a745
    style EM fill:#d4edda,stroke:#28a745
    style E1 fill:#fff3cd,stroke:#ffc107
    style E2 fill:#fff3cd,stroke:#ffc107
    style E3 fill:#fff3cd,stroke:#ffc107
    style E4 fill:#fff3cd,stroke:#ffc107
    style E5 fill:#fff3cd,stroke:#ffc107
    style S1 fill:#fff3cd,stroke:#ffc107
    style S2 fill:#cce5ff,stroke:#004085

Key principles: - Isolated stream: System events stored in $wb-system stream (separate from domain events) - Opt-in per host: Enable only the system events you need per service - Same infrastructure: System events use events, perspectives, and lenses - LocalOnly by default: No transport publishing to avoid duplicate auditing - Self-audit prevention: System events marked with [AuditEvent(Exclude = true)]

---

Quick Start

Enable System Events

Enable System Events

// In Program.cs - enable the system events you need
// (call AFTER your IEventStore/storage is configured)
services.AddSystemEvents(options => {
  // Enable event and command auditing
  options.EnableAudit();

  // Or enable specific categories
  options.EnableEventAudit();
  options.EnableCommandAudit();
  options.EnablePerspectiveEvents();
  options.EnableErrorEvents();

  // Or enable everything
  options.EnableAll();

  // LocalOnly is true by default - system events stay local
  // For centralized monitoring, use Broadcast()
  // options.Broadcast();
});

Consume System Events

System events are consumed like domain events - create perspectives:

Consume System Events

using Whizbang.Core.Perspectives;
using Whizbang.Core.SystemEvents;
using Whizbang.Core.Audit;

/// <summary>
/// Perspective that captures EventAudited system events.
/// </summary>
public sealed class AuditPerspective : IPerspectiveFor<AuditLogEntry, EventAudited> {
  public AuditLogEntry Apply(AuditLogEntry current, EventAudited @event) {
    return new AuditLogEntry {
      Id = @event.Id,
      StreamId = @event.OriginalStreamId,
      StreamPosition = @event.OriginalStreamPosition,
      EventType = @event.OriginalEventType,
      Timestamp = @event.Timestamp,
      TenantId = @event.TenantId,
      UserId = @event.UserId,
      Body = @event.OriginalBody
    };
  }
}

/// <summary>
/// Perspective that captures security events.
/// </summary>
public sealed class SecurityAuditPerspective :
    IPerspectiveFor<SecurityAuditEntry, AccessDenied>,
    IPerspectiveFor<SecurityAuditEntry, AccessGranted> {

  public SecurityAuditEntry Apply(SecurityAuditEntry current, AccessDenied @event) {
    return new SecurityAuditEntry {
      Id = @event.Id,
      EventType = "AccessDenied",
      ResourceType = @event.ResourceType,
      ResourceId = @event.ResourceId,
      UserId = @event.Scope.UserId,
      TenantId = @event.Scope.TenantId,
      Timestamp = @event.Timestamp,
      Details = new {
        RequiredPermission = @event.RequiredPermission.ToString(),
        CallerPermissions = @event.CallerPermissions.Select(p => p.ToString()).ToList(),
        Reason = @event.Reason.ToString()
      }
    };
  }

  public SecurityAuditEntry Apply(SecurityAuditEntry current, AccessGranted @event) {
    return new SecurityAuditEntry {
      Id = @event.Id,
      EventType = "AccessGranted",
      ResourceType = @event.ResourceType,
      ResourceId = @event.ResourceId,
      UserId = @event.Scope.UserId,
      TenantId = @event.Scope.TenantId,
      Timestamp = @event.Timestamp,
      Details = new {
        UsedPermission = @event.UsedPermission.ToString(),
        AccessFilter = @event.AccessFilter.ToString()
      }
    };
  }
}

---

Built-in System Events

EventAudited

Emitted when a domain event is appended to a stream (when EnableEventAudit() is configured).

EventAudited

[AuditEvent(Exclude = true, Reason = "System event - prevents infinite self-auditing loop")]
public sealed record EventAudited : ISystemEvent {
  /// <summary>
  /// Unique identifier for this audit event.
  /// </summary>
  [StreamId]
  public required Guid Id { get; init; }

  /// <summary>
  /// Event ID of the original domain event.
  /// </summary>
  public Guid OriginalEventId { get; init; }

  /// <summary>
  /// Type name of the original domain event (e.g., "OrderCreated").
  /// </summary>
  public required string OriginalEventType { get; init; }

  /// <summary>
  /// Stream ID where the original event was appended.
  /// </summary>
  public required string OriginalStreamId { get; init; }

  /// <summary>
  /// Position within the stream where the original event was appended.
  /// </summary>
  public required long OriginalStreamPosition { get; init; }

  /// <summary>
  /// Full body of the original event as JSON.
  /// </summary>
  public required JsonElement OriginalBody { get; init; }

  /// <summary>
  /// When the original event was recorded.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }

  /// <summary>
  /// Tenant identifier from event scope.
  /// </summary>
  public string? TenantId { get; init; }

  /// <summary>
  /// User identifier from event scope.
  /// </summary>
  public string? UserId { get; init; }

  /// <summary>
  /// Correlation ID for distributed tracing.
  /// </summary>
  public string? CorrelationId { get; init; }

  /// <summary>
  /// Causation ID (the message that caused the original event).
  /// </summary>
  public string? CausationId { get; init; }

  /// <summary>
  /// Optional reason from [AuditEvent(Reason = ...)] on the event type.
  /// </summary>
  public string? AuditReason { get; init; }

  /// <summary>
  /// Audit severity level. Defaults to AuditLevel.Info.
  /// </summary>
  public AuditLevel AuditLevel { get; init; } = AuditLevel.Info;

  /// <summary>
  /// Generic scope dictionary containing all security context values.
  /// Enables flexible row-level security beyond TenantId/UserId.
  /// </summary>
  public IReadOnlyDictionary<string, string?>? Scope { get; init; }
}

Use cases: - Compliance audit trails (GDPR, SOX, HIPAA) - "Who changed what, when?" queries - Event replay and debugging - Multi-tenant data access auditing

Excluding events from audit:

EventAudited - ServiceHeartbeat

// Exclude high-frequency or non-essential events from audit
[AuditEvent(Exclude = true, Reason = "High-frequency heartbeat event")]
public sealed record ServiceHeartbeat : IEvent {
  public required Guid ServiceId { get; init; }
  public required DateTimeOffset Timestamp { get; init; }
}

CommandAudited

Emitted when a command is processed by a receptor (when EnableCommandAudit() is configured).

CommandAudited

[AuditEvent(Exclude = true, Reason = "System event - prevents infinite self-auditing loop")]
public sealed record CommandAudited : ISystemEvent {
  /// <summary>
  /// Unique identifier for this audit entry.
  /// </summary>
  [StreamId]
  public required Guid Id { get; init; }

  /// <summary>
  /// Type name of the command (e.g., "CreateOrder").
  /// </summary>
  public required string CommandType { get; init; }

  /// <summary>
  /// JSON representation of the command body.
  /// </summary>
  public required JsonElement CommandBody { get; init; }

  /// <summary>
  /// When the command was processed.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }

  /// <summary>
  /// Tenant context from the command scope.
  /// </summary>
  public string? TenantId { get; init; }

  /// <summary>
  /// User ID from the command scope.
  /// </summary>
  public string? UserId { get; init; }

  /// <summary>
  /// Display name of the user, when available.
  /// </summary>
  public string? UserName { get; init; }

  /// <summary>
  /// Correlation ID for distributed tracing.
  /// </summary>
  public string? CorrelationId { get; init; }

  /// <summary>
  /// Causation ID (the message that caused this command).
  /// </summary>
  public string? CausationId { get; init; }

  /// <summary>
  /// Optional reason from [AuditEvent(Reason = ...)] on the command type.
  /// </summary>
  public string? AuditReason { get; init; }

  /// <summary>
  /// Audit severity level. Defaults to AuditLevel.Info.
  /// </summary>
  public AuditLevel AuditLevel { get; init; } = AuditLevel.Info;

  /// <summary>
  /// Name of the receptor that handled the command.
  /// </summary>
  public string? ReceptorName { get; init; }

  /// <summary>
  /// Type of the response returned by the receptor.
  /// </summary>
  public string? ResponseType { get; init; }

  /// <summary>
  /// Generic scope dictionary for flexible security context.
  /// </summary>
  public IReadOnlyDictionary<string, string?>? Scope { get; init; }
}

Use cases: - Command execution auditing - "Who executed what command?" queries - API call tracking - Performance and usage analytics

ScopeContextEstablished

Emitted when a scope context is established for a request/operation.

ScopeContextEstablished

public sealed record ScopeContextEstablished : ISystemEvent {
  [StreamId]
  public Guid Id { get; init; } = TrackedGuid.NewMedo();

  /// <summary>
  /// The established scope (TenantId, UserId, etc.).
  /// </summary>
  public required PerspectiveScope Scope { get; init; }

  /// <summary>
  /// Roles in the context.
  /// </summary>
  public required IReadOnlySet<string> Roles { get; init; }

  /// <summary>
  /// Permissions in the context.
  /// </summary>
  public required IReadOnlySet<Permission> Permissions { get; init; }

  /// <summary>
  /// Source of the context (JWT, API Key, etc.).
  /// </summary>
  public required string Source { get; init; }

  /// <summary>
  /// When the context was established.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }
}

Use cases: - Authentication audit trails - User session tracking - Security context debugging

PermissionChanged

Emitted when a user's permissions or roles change.

PermissionChanged

public sealed record PermissionChanged : ISystemEvent {
  [StreamId]
  public Guid Id { get; init; } = TrackedGuid.NewMedo();

  /// <summary>
  /// User whose permissions changed.
  /// </summary>
  public required string UserId { get; init; }

  /// <summary>
  /// Tenant context.
  /// </summary>
  public required string TenantId { get; init; }

  /// <summary>
  /// Type of change (RolesAdded, RolesRemoved, etc.).
  /// </summary>
  public required PermissionChangeType ChangeType { get; init; }

  /// <summary>
  /// Roles added (if any).
  /// </summary>
  public IReadOnlySet<string>? RolesAdded { get; init; }

  /// <summary>
  /// Roles removed (if any).
  /// </summary>
  public IReadOnlySet<string>? RolesRemoved { get; init; }

  /// <summary>
  /// Permissions added (if any).
  /// </summary>
  public IReadOnlySet<Permission>? PermissionsAdded { get; init; }

  /// <summary>
  /// Permissions removed (if any).
  /// </summary>
  public IReadOnlySet<Permission>? PermissionsRemoved { get; init; }

  /// <summary>
  /// Who made the change.
  /// </summary>
  public required string ChangedBy { get; init; }

  /// <summary>
  /// When the change occurred.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }
}

Use cases: - Role/permission change auditing - Security compliance tracking - Access control debugging

AccessGranted

Emitted when access to a sensitive resource is granted.

AccessGranted

public sealed record AccessGranted : ISystemEvent {
  [StreamId]
  public Guid Id { get; init; } = TrackedGuid.NewMedo();

  /// <summary>
  /// Type of resource access was granted to.
  /// </summary>
  public required string ResourceType { get; init; }

  /// <summary>
  /// Optional resource identifier.
  /// </summary>
  public string? ResourceId { get; init; }

  /// <summary>
  /// The permission that was used.
  /// </summary>
  public required Permission UsedPermission { get; init; }

  /// <summary>
  /// Access filter applied (e.g., tenant-scoped). ScopeFilters is a
  /// combinable [Flags] enum (None, Tenant, Organization, Customer, User, Principal).
  /// </summary>
  public required ScopeFilters AccessFilter { get; init; }

  /// <summary>
  /// Scope context at time of access.
  /// </summary>
  public required PerspectiveScope Scope { get; init; }

  /// <summary>
  /// When access was granted.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }
}

Use cases: - Privileged access auditing - Compliance reporting (who accessed what) - Security monitoring

AccessDenied

Emitted when access to a resource is denied due to insufficient permissions.

AccessDenied

public sealed record AccessDenied : ISystemEvent {
  [StreamId]
  public Guid Id { get; init; } = TrackedGuid.NewMedo();

  /// <summary>
  /// Type of resource access was denied to.
  /// </summary>
  public required string ResourceType { get; init; }

  /// <summary>
  /// Optional resource identifier.
  /// </summary>
  public string? ResourceId { get; init; }

  /// <summary>
  /// The permission that was required.
  /// </summary>
  public required Permission RequiredPermission { get; init; }

  /// <summary>
  /// Permissions the caller had.
  /// </summary>
  public required IReadOnlySet<Permission> CallerPermissions { get; init; }

  /// <summary>
  /// Roles the caller had.
  /// </summary>
  public required IReadOnlySet<string> CallerRoles { get; init; }

  /// <summary>
  /// Scope context at time of denial.
  /// </summary>
  public required PerspectiveScope Scope { get; init; }

  /// <summary>
  /// Reason for denial.
  /// </summary>
  public required AccessDenialReason Reason { get; init; }

  /// <summary>
  /// When access was denied.
  /// </summary>
  public required DateTimeOffset Timestamp { get; init; }
}

Use cases: - Security threat detection - Failed access attempt auditing - Authorization debugging

---

System Event Emitter

The ISystemEventEmitter is responsible for emitting system events to the dedicated $wb-system stream.

System Event Emitter

public interface ISystemEventEmitter {
  /// <summary>
  /// Emits an EventAudited system event for a domain event.
  /// </summary>
  Task EmitEventAuditedAsync<TEvent>(
      Guid streamId,
      long streamPosition,
      MessageEnvelope<TEvent> envelope,
      CancellationToken cancellationToken = default);

  /// <summary>
  /// Emits a CommandAudited system event for a command.
  /// </summary>
  Task EmitCommandAuditedAsync<TCommand, TResponse>(
      TCommand command,
      TResponse response,
      string receptorName,
      IMessageContext? context,
      CancellationToken cancellationToken = default) where TCommand : notnull;

  /// <summary>
  /// Emits a generic system event to the system stream.
  /// </summary>
  Task EmitAsync<TSystemEvent>(
      TSystemEvent systemEvent,
      CancellationToken cancellationToken = default) where TSystemEvent : ISystemEvent;

  /// <summary>
  /// Checks if the given type should be excluded from auditing.
  /// </summary>
  bool ShouldExcludeFromAudit(Type type);
}

Implementations: SystemEventEmitter (real) and NullSystemEventEmitter (no-op)

System Event Emitter - SystemEventEmitter

public sealed class SystemEventEmitter(
    IOptions<SystemEventOptions> options,
    IEventStore systemEventStore) : ISystemEventEmitter {
  // Emits system events only when enabled via options
  // Respects [AuditEvent(Exclude = true)] to prevent infinite loops
  // Serializes payloads in AOT-compatible way (JsonContextRegistry)
}

AddWhizbang() registers NullSystemEventEmitter as the default ISystemEventEmitter (every emit is a no-op) so injecting the emitter is always safe. Note that the built-in event audit path does not go through the emitter at all — it uses the auditing decorator and the deferred outbox channel. The emitter is used by the command-audit pipeline behavior and for manual emission.

Manual emission (advanced scenarios):

System Event Emitter - MySecurityService

public class MySecurityService {
  private readonly ISystemEventEmitter _emitter;

  public async Task GrantAccessAsync(string userId, string resourceId) {
    // ... grant access logic ...

    // Manually emit security system event
    await _emitter.EmitAsync(new AccessGranted {
      Id = TrackedGuid.NewMedo(),
      ResourceType = "SensitiveDocument",
      ResourceId = resourceId,
      UsedPermission = Permission.Read("documents"),  // "documents:read"
      AccessFilter = ScopeFilters.Tenant,             // tenant-scoped access
      Scope = new PerspectiveScope {
        TenantId = "tenant-123",
        UserId = userId
      },
      Timestamp = DateTimeOffset.UtcNow
    });
  }
}

---

System Event Configuration

System events are configured via SystemEventOptions:

System Event Configuration

public sealed class SystemEventOptions {
  /// <summary>
  /// When true, system events stay local (no transport publishing).
  /// Default is true.
  /// </summary>
  public bool LocalOnly { get; set; } = true;

  /// <summary>
  /// Controls which events are audited when event audit is enabled:
  /// AuditMode.OptOut (default) audits all events unless [AuditEvent(Exclude = true)];
  /// AuditMode.OptIn audits only events explicitly marked with [AuditEvent].
  /// </summary>
  public AuditMode AuditMode { get; set; } = AuditMode.OptOut;

  /// <summary>
  /// Enables EventAudited system events.
  /// </summary>
  public bool EventAuditEnabled { get; private set; }

  /// <summary>
  /// Enables CommandAudited system events.
  /// </summary>
  public bool CommandAuditEnabled { get; private set; }

  /// <summary>
  /// Returns true if either event or command auditing is enabled.
  /// </summary>
  public bool AuditEnabled => EventAuditEnabled || CommandAuditEnabled;

  /// <summary>
  /// Enables perspective-related system events.
  /// </summary>
  public bool PerspectiveEventsEnabled { get; private set; }

  /// <summary>
  /// Enables error-related system events.
  /// </summary>
  public bool ErrorEventsEnabled { get; private set; }
}

Configuration methods:

System Event Configuration (2)

// Inside services.AddSystemEvents(options => { ... }) - all methods are fluent

// Enable all system events
options.EnableAll();

// Enable audit (both events and commands)
options.EnableAudit();

// Enable specific categories
options.EnableEventAudit();
options.EnableCommandAudit();
options.EnablePerspectiveEvents();
options.EnableErrorEvents();

// Only audit explicitly marked events
options.AuditMode = AuditMode.OptIn;

// Broadcast system events to transport (advanced)
options.Broadcast(); // Sets LocalOnly = false

LocalOnly vs Broadcast:

System Event Configuration (3)

// Default: LocalOnly = true
// Each service audits what it processes locally
// No network traffic, no duplication
services.AddSystemEvents(options => {
  options.EnableAudit();
  // LocalOnly is true by default
});

// Broadcast mode: LocalOnly = false
// System events published to transport
// Use when you have centralized monitoring service
services.AddSystemEvents(options => {
  options.EnableAll();
  options.Broadcast(); // Sets LocalOnly = false
});

Why LocalOnly by default?

Consider this scenario: - BFF receives events from Orders and Users services - Both BFF and Users service have audit enabled

Without LocalOnly: 1. Users service audits UserCreated locally 2. Users service publishes EventAudited to transport 3. BFF receives UserCreated and audits it locally 4. BFF receives EventAudited from Users service (duplicate!)

With LocalOnly = true: 1. Users service audits UserCreated locally (stays local) 2. BFF receives UserCreated and audits it locally (stays local) 3. No duplicate audit events!

---

Transport Filtering

The SystemEventTransportFilter implements ITransportPublishFilter to control which events flow through the transport layer:

Transport Filtering

public sealed class SystemEventTransportFilter(IOptions<SystemEventOptions> options)
    : ITransportPublishFilter {
  private readonly SystemEventOptions _options = options.Value;

  public bool ShouldPublishToTransport(object message) {
    // Domain events always publish
    if (message is not ISystemEvent) {
      return true;
    }

    // System events respect LocalOnly setting
    return !_options.LocalOnly;
  }

  public bool ShouldReceiveFromTransport(Type messageType) {
    // Domain events always received
    if (!typeof(ISystemEvent).IsAssignableFrom(messageType)) {
      return true;
    }

    // System events respect LocalOnly setting
    return !_options.LocalOnly;
  }
}

Routing rules: - Domain events: Always flow through transport (cross-service communication) - System events: Respect LocalOnly setting (default: stay local)

This ensures: - Domain events drive business workflows across services - System events provide local observability without network overhead - No duplicate auditing when multiple services enable audit

---

Event Auditing Decorator

The AuditingEventStoreDecorator wraps your IEventStore implementation. On each append it builds an EventAudited envelope and queues it to the deferred outbox channel with the dedicated audit topic destination "whizbang.core.auditevents" — it deliberately does NOT depend on IEventStore or ISystemEventEmitter itself, which avoids circular DI:

Event Auditing Decorator

public sealed class AuditingEventStoreDecorator(
    IEventStore inner,
    IDeferredOutboxChannel outboxChannel,
    IOptions<SystemEventOptions> options) : IEventStore {

  // Dedicated audit topic destination for outbox messages
  public const string AUDIT_TOPIC_DESTINATION = "whizbang.core.auditevents";

  public async Task AppendAsync<TMessage>(
      Guid streamId,
      MessageEnvelope<TMessage> envelope,
      CancellationToken cancellationToken = default) {
    // First, append to the inner store
    await _inner.AppendAsync(streamId, envelope, cancellationToken);

    // Then queue an EventAudited to the deferred outbox channel
    // (respects AuditMode and [AuditEvent(Exclude = true)])
    await _emitAuditIfEligibleAsync(streamId, envelope, cancellationToken);
  }

  // ... other IEventStore methods delegate to _inner ...
}

Registration: AddSystemEvents decorates an already-registered IEventStore automatically when EnableEventAudit() is set. You can also decorate explicitly:

Event Auditing Decorator (2)

// Automatic: IEventStore registered BEFORE AddSystemEvents gets decorated
services.AddSingleton<IEventStore, PostgresEventStore>();
services.AddSystemEvents(options => options.EnableEventAudit());

// Explicit: decorate an existing IEventStore registration yourself
services.DecorateEventStoreWithAuditing();

Or use the combined method:

Event Auditing Decorator (3)

services
  .AddWhizbang()
  .WithEFCore<MyDbContext>()
  .WithDriver.Postgres;

// Add auditing AFTER storage is configured
services.AddSystemEventAuditing(options => {
  options.EnableEventAudit();
  options.EnableCommandAudit();
});

---

Command Auditing Pipeline Behavior

The CommandAuditPipelineBehavior<TCommand, TResponse> automatically emits CommandAudited system events for commands processed by receptors:

Command Auditing Pipeline Behavior

public sealed class CommandAuditPipelineBehavior<TCommand, TResponse> : PipelineBehavior<TCommand, TResponse>
    where TCommand : notnull {
  private readonly ISystemEventEmitter _emitter;
  private readonly SystemEventOptions _options;
  private readonly IMessageContext? _context;

  public override async Task<TResponse> HandleAsync(
      TCommand request,
      Func<Task<TResponse>> continuation,
      CancellationToken cancellationToken = default) {
    // Execute the next behavior or handler
    var response = await ExecuteNextAsync(continuation);

    // Check if command auditing is enabled
    if (!_options.CommandAuditEnabled) {
      return response;
    }

    // Check if this command type should be excluded from audit
    if (_emitter.ShouldExcludeFromAudit(typeof(TCommand))) {
      return response;
    }

    // Extract receptor name from context metadata
    var receptorName = _extractReceptorName();

    // Emit the audit event
    await _emitter.EmitCommandAuditedAsync(
        request,
        response,
        receptorName,
        _context,
        cancellationToken);

    return response;
  }
}

Registration (automatic with AddSystemEventAuditing):

Command Auditing Pipeline Behavior (2)

services.AddSystemEventAuditing(options => {
  options.EnableCommandAudit();
});

The pipeline behavior is registered automatically:

Command Auditing Pipeline Behavior (3)

services.TryAddSingleton(
    typeof(IPipelineBehavior<,>),
    typeof(CommandAuditPipelineBehavior<,>));

---

System Event Stream

System events are stored in a dedicated stream with a fixed identifier:

System Event Stream

public static class SystemEventStreams {
  /// <summary>
  /// The name of the dedicated system event stream.
  /// Uses $ prefix following EventStoreDB convention for system streams.
  /// </summary>
  public static string Name => "$wb-system";

  /// <summary>
  /// Stream prefix for system events.
  /// </summary>
  public static string Prefix => "$wb-";

  /// <summary>
  /// Well-known GUID for the system event stream.
  /// Fixed: 00000000-0000-0000-0000-000000000001
  /// </summary>
  public static Guid StreamId { get; } = new Guid("00000000-0000-0000-0000-000000000001");
}

Why a dedicated stream? - Isolation: System events separate from domain events - Performance: Query system events without scanning domain streams - Clarity: Clear separation of concerns - Convention: Follows EventStoreDB's $ prefix for system streams

---

Registration and Setup

Basic Registration

Basic Registration

services.AddSystemEvents(options => {
  options.EnableAudit();
});

This registers: - ITransportPublishFilter (→ SystemEventTransportFilter) for transport filtering - The AuditingEventStoreDecorator around an already-registered IEventStore, when EnableEventAudit() is set

(ISystemEventEmitter itself is registered by AddWhizbang(), defaulting to the no-op NullSystemEventEmitter.)

Full Auditing Registration

Full Auditing Registration

services.AddSystemEventAuditing(options => {
  options.EnableEventAudit();
  options.EnableCommandAudit();
});

This registers: - All basic system event services (everything AddSystemEvents does) - CommandAuditPipelineBehavior<,> for command auditing

Complete Setup Example

Complete Setup Example

// In Program.cs
var builder = WebApplication.CreateBuilder(args);

// Configure storage
builder.Services
  .AddWhizbang()
  .WithEFCore<MyDbContext>()
  .WithDriver.Postgres;

// Add system event auditing AFTER storage is configured
builder.Services.AddSystemEventAuditing(options => {
  options.EnableAll(); // Enable all system events
  // LocalOnly = true by default (no transport publishing)
});

// Perspectives (including ones that consume system events) are discovered
// by the source generator; register them via the generated extension:
builder.Services.AddWhizbangPerspectives();

var app = builder.Build();
app.Run();

---

Best Practices

1. Enable Only What You Need

Enable Only What You Need

// BFF: Enable full audit for compliance
services.AddSystemEvents(options => {
  options.EnableAudit();
});

// Background worker: Maybe just errors
services.AddSystemEvents(options => {
  options.EnableErrorEvents();
});

// Read-only query service: No system events needed
services.AddWhizbang();

2. Use LocalOnly (Default)

Use LocalOnly (Default)

// Default behavior - system events stay local
services.AddSystemEvents(options => {
  options.EnableAudit();
  // LocalOnly = true by default
});

// Each service maintains its own audit trail
// No network traffic for system events
// No duplicate auditing

3. Exclude High-Frequency Events

Exclude High-Frequency Events

// Exclude events that would create excessive audit volume
[AuditEvent(Exclude = true, Reason = "High-frequency telemetry event")]
public sealed record MetricCaptured : IEvent {
  public required string MetricName { get; init; }
  public required double Value { get; init; }
  public required DateTimeOffset Timestamp { get; init; }
}

4. Prevent Self-Auditing Loops

System events are already marked with [AuditEvent(Exclude = true)] to prevent infinite loops:

Prevent Self-Auditing Loops

[AuditEvent(Exclude = true, Reason = "System event - prevents infinite self-auditing loop")]
public sealed record EventAudited : ISystemEvent {
  // ...
}

Never remove this attribute from system events!

5. Query System Events Like Domain Events

Query System Events Like Domain Events

public class SecurityService {
  private readonly ILensQuery<SecurityAuditEntry> _securityLens;

  public async Task<IReadOnlyList<SecurityAuditEntry>> GetFailedAccessAttemptsAsync(
      string userId,
      DateTimeOffset since,
      CancellationToken ct) {
    return await _securityLens.DefaultScope.Query
        .Where(r => r.Data.EventType == "AccessDenied" &&
                    r.Data.UserId == userId &&
                    r.Data.Timestamp >= since)
        .OrderByDescending(r => r.Data.Timestamp)
        .Select(r => r.Data)
        .ToListAsync(ct);
  }
}

---

Common Patterns

Centralized Monitoring Service

Centralized Monitoring Service

// Monitoring service receives system events from all hosts
services.AddSystemEvents(options => {
  options.EnableAll();
  options.Broadcast(); // Receive from transport
});

// All other services use LocalOnly (default)
// They emit system events but don't broadcast them

Selective Security Auditing

Selective Security Auditing

// Only audit high-sensitivity operations
public class DocumentService {
  private readonly ISystemEventEmitter _emitter;

  public async Task<Document> ViewDocumentAsync(Guid documentId, SecurityContext ctx) {
    var doc = await _repo.GetAsync(documentId);

    // Emit AccessGranted for high-sensitivity documents only
    if (doc.Sensitivity == Sensitivity.High) {
      await _emitter.EmitAsync(new AccessGranted {
        Id = TrackedGuid.NewMedo(),
        ResourceType = "Document",
        ResourceId = documentId.ToString(),
        UsedPermission = Permission.Read("documents"),
        AccessFilter = ScopeFilters.Tenant,
        Scope = new PerspectiveScope {
          TenantId = ctx.TenantId,
          UserId = ctx.UserId
        },
        Timestamp = DateTimeOffset.UtcNow
      });
    }

    return doc;
  }
}

Multi-Tenant Audit Queries

Multi-Tenant Audit Queries

public class AuditService {
  private readonly ILensQuery<AuditLogEntry> _auditLens;

  // Tenant admin views their own audit trail
  public async Task<IReadOnlyList<AuditLogEntry>> GetTenantAuditTrailAsync(
      string tenantId,
      CancellationToken ct) {
    return await _auditLens.DefaultScope.Query
        .Where(r => r.Data.TenantId == tenantId)
        .OrderByDescending(r => r.Data.Timestamp)
        .Select(r => r.Data)
        .ToListAsync(ct);
  }

  // System admin views cross-tenant audit trail
  public async Task<IReadOnlyList<AuditLogEntry>> GetSystemAuditTrailAsync(
      DateTimeOffset since,
      CancellationToken ct) {
    return await _auditLens.DefaultScope.Query
        .Where(r => r.Data.Timestamp >= since)
        .OrderByDescending(r => r.Data.Timestamp)
        .Select(r => r.Data)
        .ToListAsync(ct);
  }
}

---

Related Documentation

• Audit Logging - Compliance-ready audit logging using system events
• Message Security - Security context and permissions
• Perspectives - Consuming system events with perspectives
• Event Store - Storage infrastructure for system events
• Event Streams - Stream concepts and conventions

---

Summary

System events provide observability, auditing, and diagnostics for Whizbang infrastructure:

• Isolated in $wb-system stream - separate from domain events
• Opt-in per host - enable only what you need per service
• LocalOnly by default - no transport publishing, no duplication
• Same infrastructure - consumed via perspectives and lenses
• Self-audit prevention - system events excluded from audit
• Built-in events - EventAudited, CommandAudited, security events, perspective rebuild/rewind events, migration events
• Extensible - emit custom system events for your scenarios

Use system events to build compliance-ready audit trails, security monitoring, and operational insights without polluting your domain model.

---

Perspective Rebuild Events

These events are emitted during perspective rebuild operations (enabled via EnablePerspectiveEvents()). They track the full lifecycle of a rebuild.

PerspectiveRebuildStarted

Emitted when a perspective rebuild starts (any mode).

PerspectiveRebuildStarted

public record PerspectiveRebuildStarted(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    RebuildMode Mode,
    int TotalStreams,
    DateTimeOffset StartedAt
) : IEvent;

PerspectiveRebuildProgress

Emitted periodically during a rebuild to report progress.

PerspectiveRebuildProgress

public record PerspectiveRebuildProgress(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    RebuildMode Mode,
    int ProcessedStreams,
    int TotalStreams,
    int EventsReplayed,
    DateTimeOffset StartedAt
) : IEvent;

PerspectiveRebuildCompleted

Emitted when a perspective rebuild completes successfully.

PerspectiveRebuildCompleted

public record PerspectiveRebuildCompleted(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    RebuildMode Mode,
    int StreamsProcessed,
    int EventsReplayed,
    TimeSpan Duration
) : IEvent;

PerspectiveRebuildFailed

Emitted when a perspective rebuild fails.

PerspectiveRebuildFailed

public record PerspectiveRebuildFailed(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    RebuildMode Mode,
    string Error,
    int StreamsProcessedBeforeFailure,
    TimeSpan Duration
) : IEvent;

---

Perspective Rewind Events

These events are emitted when a perspective rewinds due to a late-arriving event. Rewind replays events from the nearest snapshot to incorporate out-of-order events.

PerspectiveRewindStarted

Emitted when a perspective rewind begins.

PerspectiveRewindStarted

public record PerspectiveRewindStarted(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    Guid TriggeringEventId,
    Guid? ReplayFromSnapshotEventId,
    bool HasSnapshot,
    DateTimeOffset StartedAt
) : IEvent;

PerspectiveRewindCompleted

Emitted when a perspective rewind completes successfully.

PerspectiveRewindCompleted

public record PerspectiveRewindCompleted(
    [property: StreamId] Guid StreamId,
    string PerspectiveName,
    Guid TriggeringEventId,
    Guid FinalEventId,
    int EventsReplayed,
    DateTimeOffset StartedAt,
    DateTimeOffset CompletedAt
) : IEvent;

StreamRewindStarted / StreamRewindCompleted

Stream-level bracket events: StreamRewindStarted is emitted once per stream before any per-perspective rewinds begin, and StreamRewindCompleted once after all of them finish.

StreamRewindStarted / StreamRewindCompleted

public record StreamRewindStarted(
    [property: StreamId] Guid StreamId,
    string[] PerspectiveNames,
    Guid TriggerEventId,
    DateTimeOffset StartedAt
) : IEvent;

public record StreamRewindCompleted(
    [property: StreamId] Guid StreamId,
    string[] PerspectiveNames,
    int TotalEventsReplayed,
    DateTimeOffset StartedAt,
    DateTimeOffset CompletedAt
) : IEvent;

---

Migration Events

These events track the lifecycle of database migrations. They are emitted during AddWhizbang() startup when migrations are applied.

MigrationItemStarted

Emitted when an individual migration starts processing.

MigrationItemStarted

public record MigrationItemStarted(
    [property: StreamId] Guid StreamId,
    string MigrationKey,
    MigrationStrategy Strategy,
    string? OldHash,
    string NewHash
) : IEvent;

MigrationItemCompleted

Emitted when an individual migration completes.

MigrationItemCompleted

public record MigrationItemCompleted(
    [property: StreamId] Guid StreamId,
    string MigrationKey,
    MigrationStatus Status,
    string StatusDescription,
    TimeSpan Duration
) : IEvent;

MigrationItemFailed

Emitted when an individual migration fails.

MigrationItemFailed

public record MigrationItemFailed(
    [property: StreamId] Guid StreamId,
    string MigrationKey,
    MigrationStatus Status,
    MigrationFailureReason FailureReason,
    string Error,
    TimeSpan Duration
) : IEvent;

MigrationBatchStarted

Emitted when the full migration batch starts (all infrastructure + perspectives).

MigrationBatchStarted

public record MigrationBatchStarted(
    [property: StreamId] Guid StreamId,
    string LibraryVersion,
    int TotalMigrations,
    int TotalPerspectives
) : IEvent;

MigrationBatchCompleted

Emitted when the full migration batch completes, including per-item results.

MigrationBatchCompleted

public record MigrationBatchCompleted(
    [property: StreamId] Guid StreamId,
    string LibraryVersion,
    MigrationBatchItemResult[] Results,
    int Applied,
    int Updated,
    int Skipped,
    int Failed,
    TimeSpan TotalDuration
) : IEvent;

public record MigrationBatchItemResult(
    string MigrationKey,
    MigrationStatus Status,
    string StatusDescription);

Migration Enums

MigrationStatus

Status of a migration item in wh_schema_migrations:

Value	Description
Applied (1)	Migration was applied for the first time
Updated (2)	Migration was updated (hash changed)
Skipped (3)	Migration was skipped (hash unchanged)
MigratingInBackground (4)	Migration is running in the background
Failed (-1)	Migration failed

MigrationStrategy

Strategy used for executing a migration:

Value	Description
DirectDdl	Direct DDL execution (CREATE TABLE, ALTER, etc.)
ColumnCopy	Column copy strategy for zero-downtime changes
EventReplay	Event replay strategy for perspective migrations

MigrationFailureReason

Reason a migration failed:

Value	Description
Unknown (0)	Unknown failure reason
SqlError (1)	SQL execution error
Timeout (2)	Migration timed out
ColumnTypeMismatch (3)	Column type mismatch during copy
DataCopyFailed (4)	Data copy operation failed
SwapFailed (5)	Column swap operation failed