.NET SDK

Hapnd .NET SDK

Official .NET SDK for Hapnd, the event sourcing platform built for engineers. Append events, query aggregate state, subscribe to real-time projection updates, and build event-sourced systems with a fluent type-safe API, built-in resilience via Polly, ASP.NET Core dependency injection, and distributed tracing support.

Installation

dotnet add package Hapnd.Sdk

Requires .NET 10.0.

Quick Start

using Hapnd.Client;

var hapnd = new HapndClient("sk_live_your_api_key");

// Append an event
var result = await hapnd.Aggregate("order_001")
    .Append(new OrderCreated("customer_123"));

Console.WriteLine($"EventId: {result.EventId}, Version: {result.Version}");

// Check if state was computed by a bound reducer
if (result.HasState)
{
    var state = result.State<OrderState>();
    Console.WriteLine($"Order total: {state.Total}");
}

Creating a Client

Direct Instantiation

// Minimal, uses shared static HttpClient with connection pooling
var hapnd = new HapndClient("sk_live_your_api_key");

// With full configuration
var hapnd = new HapndClient(new HapndClientOptions
{
    ApiKey = "sk_live_your_api_key",
    BaseUrl = "https://hapnd-api.lightestnight.workers.dev",
    Timeout = TimeSpan.FromSeconds(60),
    Resilience = new HapndResilienceOptions
    {
        MaxRetryAttempts = 5
    }
});

The client is thread-safe and should be used as a singleton. It uses a shared static HttpClient with SocketsHttpHandler and a 2-minute pooled connection lifetime to respect DNS changes while maintaining connection reuse.

Dependency Injection

// Simple form
services.AddHapnd("sk_live_your_api_key");

// Full configuration
services.AddHapnd(options =>
{
    options.ApiKey = configuration["Hapnd:ApiKey"];
    options.Timeout = TimeSpan.FromSeconds(60);
    options.Resilience = new HapndResilienceOptions
    {
        MaxRetryAttempts = 5,
        EnableCircuitBreaker = true
    };
});

Then inject IHapndClient anywhere in your application:

public class OrderService(IHapndClient hapnd)
{
    public async Task PlaceOrder(string customerId)
    {
        var result = await hapnd.Aggregate($"order_{Guid.NewGuid()}")
            .Append(new OrderPlaced(customerId));
    }
}

The DI registration configures IHapndClient as a singleton with a named HttpClient via IHttpClientFactory for proper HTTP connection lifecycle management.

Configuration Options

Option	Type	Default	Description
`ApiKey`	`string`	(required)	Your Hapnd API key
`BaseUrl`	`string`	`https://hapnd-api.lightestnight.workers.dev`	API base URL
`Timeout`	`TimeSpan`	30 seconds	HTTP request timeout
`Resilience`	`HapndResilienceOptions?`	Enabled with defaults	Retry and circuit breaker config. Set to `null` to disable.

Appending Events

Single Event

var result = await hapnd.Aggregate("order_123")
    .Append(new ItemAdded("Widget", 25.00m));

Console.WriteLine($"EventId: {result.EventId}");
Console.WriteLine($"AggregateId: {result.AggregateId}");
Console.WriteLine($"AggregateType: {result.AggregateType}");
Console.WriteLine($"Version: {result.Version}");
Console.WriteLine($"Timestamp: {result.Timestamp}");
Console.WriteLine($"HasState: {result.HasState}");

AppendResult properties:

Property	Type	Description
`EventId`	`string`	Unique identifier for the appended event
`AggregateId`	`string`	The aggregate this event was appended to
`AggregateType`	`string`	The type of the aggregate
`Version`	`int`	Version of the aggregate after this event
`Timestamp`	`DateTimeOffset`	Server-assigned timestamp
`HasState`	`bool`	Whether computed state is available (reducer bound)
`State<T>()`	`T?`	Deserialize the computed state to the specified type

Batch Append

var result = await hapnd.Aggregate("order_123")
    .AppendMany([
        new ItemAdded("Widget", 25.00m),
        new ItemAdded("Gadget", 15.00m)
    ]);

Console.WriteLine($"Appended {result.Events.Count} events (versions {result.StartVersion}-{result.EndVersion})");

All events in a batch are stored atomically; either all succeed or none do.

AppendManyResult properties:

Property	Type	Description
`AggregateId`	`string`	The aggregate these events were appended to
`AggregateType`	`string`	The type of the aggregate
`StartVersion`	`int`	Version after the first event in the batch
`EndVersion`	`int`	Version after the last event in the batch
`Events`	`IReadOnlyList<AppendedEvent>`	Individual results for each appended event
`HasState`	`bool`	Whether computed state is available
`State<T>()`	`T?`	Deserialize the computed state after all events

Event Type Resolution

By default, the event type is derived from the class name:

// Event type: "OrderCreated"
await hapnd.Aggregate("order_123").Append(new OrderCreated("customer_456"));

Override with the [EventType] attribute:

[EventType("order_placed")]
public class OrderPlaced
{
    public string CustomerId { get; init; }
}

// Event type: "order_placed"
await hapnd.Aggregate("order_123").Append(new OrderPlaced { CustomerId = "customer_456" });

Aggregates

Aggregate Type Inference

The SDK infers the aggregate type from the ID using the {type}_{id} or {type}-{id} convention:

Aggregate ID	Inferred Type
`order_123`	`order`
`cart-abc`	`cart`
`inventory_sku_42`	`inventory`

Override when your IDs don’t follow this convention:

await hapnd.Aggregate("ORD-2024-00123")
    .WithAggregateType("order")
    .Append(new OrderPlaced { CustomerId = "customer_456" });

Optimistic Concurrency

Use .ExpectVersion() to prevent lost updates when multiple processes modify the same aggregate:

// Read current state
var current = await hapnd.Aggregate("order_123").GetState<OrderState>();

try
{
    // Write with version check
    await hapnd.Aggregate("order_123")
        .ExpectVersion(current.Version)
        .Append(new ItemAdded("Widget", 25.00m));
}
catch (HapndConcurrencyException ex)
{
    Console.WriteLine($"Expected version {ex.ExpectedVersion}, actual {ex.ActualVersion}");
    // Reload state and retry
}

Use ExpectVersion(0) to assert the aggregate does not yet exist:

await hapnd.Aggregate("order_new")
    .ExpectVersion(0)
    .Append(new OrderCreated("customer_789"));

Querying State

var aggregate = await hapnd.Aggregate("order_123").GetState<OrderState>();

if (aggregate is not null)
{
    Console.WriteLine($"Version: {aggregate.Version}");
    Console.WriteLine($"Total: {aggregate.State.Total}");
    Console.WriteLine($"Last Modified: {aggregate.LastModified}");
}

Returns AggregateState<TState>?, where null indicates the aggregate does not exist or no reducer is bound.

AggregateState<T> properties:

Property	Type	Description
`AggregateId`	`string`	The aggregate identifier
`AggregateType`	`string`	The type of the aggregate
`Version`	`int`	Current version of the aggregate
`State`	`T`	The computed state value
`LastModified`	`DateTimeOffset`	Timestamp of the last event applied

Distributed Tracing

Correlation IDs

Group related operations across services:

var correlationId = Guid.NewGuid().ToString();

await hapnd.Aggregate("order_123")
    .WithCorrelation(correlationId)
    .Append(new OrderCreated("customer_456"));

await hapnd.Aggregate("inventory_widget")
    .WithCorrelation(correlationId)
    .Append(new StockReserved("order_123", 1));

Causation IDs

Track direct cause-and-effect between events:

var orderResult = await hapnd.Aggregate("order_123")
    .Append(new OrderCreated("customer_456"));

// This event was caused by the OrderCreated event
await hapnd.Aggregate("notification_123")
    .WithCausation(orderResult.EventId)
    .Append(new OrderConfirmationSent("customer_456"));

Metadata

Attach arbitrary context to events:

await hapnd.Aggregate("order_123")
    .WithMetadata(new
    {
        UserId = "user_456",
        IpAddress = "192.168.1.1",
        UserAgent = "MyApp/1.0"
    })
    .Append(new OrderCreated("customer_456"));

Fluent Builder

All configuration methods return the builder (IAggregateBuilder) for chaining. Terminal methods execute the API call.

Configuration methods: WithAggregateType, ExpectVersion, WithCorrelation, WithCausation, WithMetadata

Terminal methods: Append, AppendMany, GetState<T>

var result = await hapnd.Aggregate("order_123")
    .WithAggregateType("order")
    .ExpectVersion(3)
    .WithCorrelation("corr_abc")
    .WithCausation("evt_previous")
    .WithMetadata(new { UserId = "user_456", Source = "web" })
    .Append(new ItemAdded("Widget", 25.00m));

Writing Reducers

Reducers compute aggregate state synchronously on every event append. Bind a reducer to an aggregate type, and the response to every event append includes the freshly computed state.

The Contracts Package

dotnet add package Hapnd.Projections.Contracts

Implementing a Reducer

using Hapnd.Projections.Contracts;

public record OrderState
{
    public decimal Total { get; init; }
    public List<string> Items { get; init; } = [];
    public int ItemCount { get; init; }
    public string? CustomerId { get; init; }
}

public class OrderReducer : IReducer<OrderState>
{
    public OrderState Apply(OrderState? state, Event @event)
    {
        state ??= new OrderState();

        return @event.Type switch
        {
            "OrderCreated" => state with
            {
                CustomerId = @event.GetData<OrderCreatedData>()?.CustomerId
            },
            "ItemAdded" => ApplyItemAdded(state, @event.GetData<ItemAddedData>()),
            _ => state
        };
    }

    private static OrderState ApplyItemAdded(OrderState state, ItemAddedData data)
    {
        return state with
        {
            Total = state.Total + data.Price,
            Items = state.Items.Append(data.Item).ToList(),
            ItemCount = state.ItemCount + 1
        };
    }
}

public record OrderCreatedData(string? CustomerId);
public record ItemAddedData(string Item, decimal Price);

Reducer Rules

Pure function - no side effects, no I/O, no network calls
No mutation - return a new state instance, do not modify the input
Idempotent - applying the same event twice produces the same result
Handle unknown events - return the current state unchanged for unrecognised event types
Roslyn static analysis enforces these constraints at compile time, blocking dangerous namespaces (System.IO, System.Net, System.Reflection, System.Diagnostics, System.Threading)

The Event Record

Property	Type	Description
`Id`	`string`	Unique event identifier
`TenantId`	`string`	Tenant identifier for multi-tenancy
`AggregateId`	`string`	Aggregate this event belongs to
`Type`	`string`	Event type discriminator
`Data`	`object`	Event payload
`Timestamp`	`long`	Event timestamp (Unix milliseconds)
`Metadata`	`Dictionary<string, object>?`	Optional metadata

Use @event.GetData<T>() for strongly-typed deserialization:

var data = @event.GetData<ItemAddedData>();
Console.WriteLine($"Added {data.Item} for {data.Price:C}");

Use @event.TryGetData<T>() for safe deserialization that returns default on failure.

Multi-Reducer DLLs

A single DLL can contain multiple IReducer<T> implementations. The platform auto-discovers all implementations at compile time. The aggregate type is inferred by stripping the “Reducer” suffix from the class name and lowercasing: OrderReducer → order, MailboxReducer → mailbox.

Override with the [AggregateType] attribute:

[AggregateType("user-account")]
public class UserReducer : IReducer<UserState>
{
    public UserState Apply(UserState? state, Event @event)
    {
        state ??= new UserState();
        return @event.Type switch
        {
            "UserRegistered" => state with { Email = @event.GetData<UserRegisteredData>().Email },
            _ => state
        };
    }
}

Deploying

# Authenticate
npx @hapnd/cli login sk_live_your_api_key

# Upload and compile
npx @hapnd/cli deploy

# Bind the reducer to an aggregate type
npx @hapnd/cli bind order red_abc123

# Check compilation status
npx @hapnd/cli status red_abc123

Writing Projections

Projections compute state asynchronously across aggregates. They implement IProjection<TState> and must define a ResolveKey method that determines how state is keyed. This is what enables cross-aggregate read models; state is keyed by whatever you return from ResolveKey, not just by aggregate ID.

using Hapnd.Projections.Contracts;

public record DashboardState
{
    public int TotalOrders { get; init; }
    public decimal Revenue { get; init; }
}

public class DashboardProjection : IProjection<DashboardState>
{
    public string? ResolveKey(Event @event)
        => @event.AggregateId;

    public DashboardState Apply(DashboardState? state, Event @event)
    {
        state ??= new DashboardState();

        return @event.Type switch
        {
            "OrderCreated" => state with { TotalOrders = state.TotalOrders + 1 },
            "ItemAdded" => state with
            {
                Revenue = state.Revenue + @event.GetData<ItemAddedData>().Price
            },
            _ => state
        };
    }
}

For cross-aggregate read models, return a different key from ResolveKey; for example, a mailbox ID extracted from event data, so events from many email aggregates all route to the same mailbox state:

public class MailboxProjection : IProjection<MailboxView>
{
    public string? ResolveKey(Event @event)
    {
        return @event.Type switch
        {
            "EmailReceived" => @event.GetData<EmailReceivedData>().MailboxId,
            "EmailRead"     => @event.GetData<EmailReadData>().MailboxId,
            _               => null  // Skip events this projection doesn't handle
        };
    }

    public MailboxView Apply(MailboxView? state, Event @event)
    {
        state ??= new MailboxView();

        return @event.Type switch
        {
            "EmailReceived" => ApplyEmailReceived(state, @event),
            "EmailRead"     => ApplyEmailRead(state, @event),
            _ => state
        };
    }

    // ... Apply methods
}

Returning null from ResolveKey skips the event entirely; no state is loaded or saved.

No IProjection.Apply(object?, Event) boilerplate needed; the contracts handle this via default interface methods.

On upload, Hapnd compiles your code, discovers IProjection<T> implementations, replays historical events through ResolveKey → Apply, and activates automatically. No checkpoints or replay logic to manage.

See the Projections concept page for the full ResolveKey explanation and more examples.

Querying Projection State

You can query the current state of a projection directly by its state key, without subscribing to changes.

var state = await hapnd.GetProjectionState<OrderSummary>("proj_abc123", "customer-123");

if (state != null)
{
    Console.WriteLine($"Total spend: {state.State.TotalSpend}");
    Console.WriteLine($"Last updated: {state.UpdatedAt}");
}

Returns null if the projection or state key doesn’t exist.

If the projection is still catching up on historical events, a ProjectionCatchingUpException is thrown. Handle it explicitly rather than polling:

try
{
    var state = await hapnd.GetProjectionState<OrderSummary>("proj_abc123", "customer-123");
}
catch (ProjectionCatchingUpException ex)
{
    Console.WriteLine($"Projection {ex.ProjectionId} is still processing historical events");
}

Real-Time Subscriptions

Basic Usage

var subscription = hapnd.Subscriptions()
    .OnStateChanged<OrderState>(async (update, ct) =>
    {
        Console.WriteLine($"Order {update.AggregateId} updated to version {update.Version}");
        Console.WriteLine($"Total: {update.State.Total}");
    })
    .OnError(async (error, ct) =>
    {
        Console.WriteLine($"Stream {error.ProjectionId} failed: {error.Exception.Message}");
        error.Action.Reconnect();
    })
    .Subscribe();

// Later, graceful shutdown:
await subscription.DisposeAsync();

Projection Attribute

Decorate your state class with [Projection] to bind it to a projection ID:

[Projection("proj_orders")]
public class OrderState
{
    public decimal Total { get; set; }
    public List<string> Items { get; set; } = [];
    public int ItemCount { get; set; }
}

// Projection ID resolved automatically from the attribute
hapnd.Subscriptions()
    .OnStateChanged<OrderState>(async (update, ct) =>
    {
        Console.WriteLine(update.State.Total);
    })
    .OnError(async (error, ct) => error.Action.Reconnect())
    .Subscribe();

If [Projection] is missing, OnStateChanged<T>() throws HapndValidationException. Use the explicit ID overload when the attribute is not present:

hapnd.Subscriptions()
    .OnStateChanged<OrderState>("proj_orders", async (update, ct) =>
    {
        Console.WriteLine(update.State.Total);
    })
    .OnError(async (error, ct) => error.Action.Reconnect())
    .Subscribe();

Compiler-Enforced Error Handling

The SDK uses a type-state pattern to enforce error handler registration at compile time:

HapndSubscriptionBuilder    →  .OnError()  →  ConfiguredSubscriptionBuilder
(no Subscribe method)                          (has Subscribe method)

// Compiles: error handler is registered before Subscribe
hapnd.Subscriptions()
    .OnStateChanged<OrderState>(async (update, ct) => { })
    .OnError(async (error, ct) => error.Action.Reconnect())
    .Subscribe();  // ConfiguredSubscriptionBuilder has Subscribe()

// Does NOT compile: HapndSubscriptionBuilder has no Subscribe method
hapnd.Subscriptions()
    .OnStateChanged<OrderState>(async (update, ct) => { })
    .Subscribe();  // Compile error!

Error Handling

StreamError properties:

Property	Type	Description
`ProjectionId`	`string`	The projection whose handler failed
`Sequence`	`long`	Last successfully processed sequence number
`Exception`	`Exception`	The exception thrown by the handler
`Action`	`StreamErrorAction`	Call exactly one method to signal how to proceed

Three recovery actions:

Action	Effect
`Action.Reconnect()`	Resume the stream from the last good sequence with backoff
`Action.Stop()`	Stop this stream only; other streams continue
`Action.Shutdown()`	Tear down the entire subscription, stopping all streams

Safeguards:

If the error handler itself throws, the failing stream stops automatically
If no action is called within 30 seconds, the stream stops automatically
Sequence advances only after the handler completes successfully; on reconnect, the server resends from the last acknowledged position

Multiple Projections

hapnd.Subscriptions()
    .OnStateChanged<OrderState>(async (update, ct) =>
    {
        Console.WriteLine($"Order: {update.State.Total}");
    })
    .OnStateChanged<InventoryState>(async (update, ct) =>
    {
        Console.WriteLine($"Inventory: {update.State.StockLevel}");
    })
    .OnError(async (error, ct) =>
    {
        Console.WriteLine($"Stream {error.ProjectionId} failed");
        error.Action.Reconnect();
    })
    .Subscribe();

DI Integration with IHostedService

Use .AddSubscriptions() to register subscriptions that start and stop with the application:

services.AddHapnd(options =>
{
    options.ApiKey = configuration["Hapnd:ApiKey"];
})
.AddSubscriptions((subs, sp) =>
    subs.OnStateChanged<OrderState>(async (update, ct) =>
    {
        await using var scope = sp.CreateAsyncScope();
        var dashboard = scope.ServiceProvider.GetRequiredService<IDashboardService>();
        await dashboard.Update(update.State);
    })
    .OnError(async (error, ct) =>
    {
        var logger = sp.GetRequiredService<ILogger<Program>>();
        logger.LogError(error.Exception, "Stream {ProjectionId} failed", error.ProjectionId);
        error.Action.Reconnect();
    })
);

ProjectionUpdate Properties

Property	Type	Description
`ProjectionId`	`string`	The projection that produced this update
`AggregateId`	`string`	The aggregate whose state changed
`AggregateType`	`string`	The aggregate type
`Version`	`int`	Aggregate version after this change
`Sequence`	`long`	Server-assigned sequence number for resumption
`State`	`TState`	The computed state
`TriggeredBy`	`string`	Event ID that triggered this state change
`Timestamp`	`DateTimeOffset`	Server timestamp of the state change

Resilience

Retry

Exponential backoff with jitter via Polly. Automatically retries transient failures:

Retried: Network errors, timeouts, 5xx server errors, 429 (rate limit), 408 (request timeout)

Not retried: 4xx client errors, 409 (concurrency conflict), cancellation

Defaults: 3 attempts, 500ms initial delay, 10s max delay

Circuit Breaker

Prevents cascading failures by temporarily stopping requests when error rates are high.

States: Closed (normal) → Open (rejecting requests) → Half-Open (testing recovery)

Defaults: Opens after 50% failure rate over 60 seconds with minimum 10 requests, stays open for 30 seconds before transitioning to half-open.

Configuration

services.AddHapnd(options =>
{
    options.ApiKey = "sk_live_your_api_key";
    options.Resilience = new HapndResilienceOptions
    {
        MaxRetryAttempts = 5,
        RetryDelay = TimeSpan.FromMilliseconds(200),
        MaxRetryDelay = TimeSpan.FromSeconds(5),
        EnableCircuitBreaker = true,
        CircuitBreakerDuration = TimeSpan.FromSeconds(30),
        CircuitBreakerSamplingDuration = TimeSpan.FromSeconds(60),
        CircuitBreakerFailureRatio = 0.5,
        CircuitBreakerMinimumThroughput = 10
    };
});

Disabling

options.Resilience = null;

Error Handling

try
{
    await hapnd.Aggregate("order_123")
        .ExpectVersion(5)
        .Append(new ItemAdded("Widget", 25.00m));
}
catch (HapndValidationException ex)
{
    // Client-side validation failure (before any network request)
    Console.WriteLine($"Validation: {ex.Message}");
}
catch (HapndConcurrencyException ex)
{
    // 409, version mismatch
    Console.WriteLine($"Expected {ex.ExpectedVersion}, actual {ex.ActualVersion}");
}
catch (HapndAggregateTypeMismatchException ex)
{
    // 400, aggregate type conflict
    Console.WriteLine($"Expected type '{ex.ExpectedType}', actual '{ex.ActualType}'");
}
catch (HapndApiException ex)
{
    // Server returned an error (varies by status code)
    Console.WriteLine($"API error {ex.StatusCode}: {ex.Message}");
}
catch (HapndNetworkException ex)
{
    // Network failure (DNS, connection refused, timeout)
    Console.WriteLine($"Network error: {ex.Message}");
}

Exception	HTTP Status	Description
`HapndValidationException`	N/A	Client-side validation failure before request
`HapndConcurrencyException`	409	Optimistic concurrency version mismatch
`HapndAggregateTypeMismatchException`	400	Aggregate type conflicts with existing events
`HapndApiException`	Varies	Server returned an error response
`HapndNetworkException`	N/A	Network-level failure (DNS, timeout, etc.)

API Endpoints

Endpoint	Method	SDK Method
`/events`	POST	`Append`
`/events/batch`	POST	`AppendMany`
`/aggregate-types/{type}/{id}/state`	GET	`GetState<T>`
`/projections/{id}/stream`	WebSocket	`Subscriptions`

Full Example

using Hapnd.Client;

public record OrderState
{
    public decimal Total { get; init; }
    public List<string> Items { get; init; } = [];
    public int ItemCount { get; init; }
    public string? CustomerId { get; init; }
}

public record OrderCreated(string CustomerId);
public record ItemAdded(string Item, decimal Price);

public class OrderService(IHapndClient hapnd)
{
    public async Task<string> CreateOrder(string customerId)
    {
        var orderId = $"order_{Guid.NewGuid()}";

        var result = await hapnd.Aggregate(orderId)
            .ExpectVersion(0)
            .WithCorrelation(Guid.NewGuid().ToString())
            .Append(new OrderCreated(customerId));

        Console.WriteLine($"Order {orderId} created at version {result.Version}");
        return orderId;
    }

    public async Task AddItem(string orderId, string item, decimal price, int expectedVersion)
    {
        try
        {
            await hapnd.Aggregate(orderId)
                .ExpectVersion(expectedVersion)
                .Append(new ItemAdded(item, price));
        }
        catch (HapndConcurrencyException)
        {
            var current = await hapnd.Aggregate(orderId).GetState<OrderState>();
            if (current is not null)
            {
                await hapnd.Aggregate(orderId)
                    .ExpectVersion(current.Version)
                    .Append(new ItemAdded(item, price));
            }
        }
    }

    public async Task<OrderState?> GetOrder(string orderId)
    {
        var aggregate = await hapnd.Aggregate(orderId).GetState<OrderState>();
        return aggregate?.State;
    }
}

// DI setup
var builder = WebApplication.CreateBuilder(args);

builder.Services.AddHapnd(options =>
{
    options.ApiKey = builder.Configuration["Hapnd:ApiKey"]!;
})
.AddSubscriptions((subs, sp) =>
    subs.OnStateChanged<OrderState>("proj_orders", async (update, ct) =>
    {
        var logger = sp.GetRequiredService<ILogger<Program>>();
        logger.LogInformation("Order {AggregateId} updated: {Total}",
            update.AggregateId, update.State.Total);
    })
    .OnError(async (error, ct) =>
    {
        var logger = sp.GetRequiredService<ILogger<Program>>();
        logger.LogError(error.Exception, "Stream {ProjectionId} failed", error.ProjectionId);
        error.Action.Reconnect();
    })
);

builder.Services.AddScoped<OrderService>();

License & Links

Licensed under Apache 2.0.