LowCodeHub.SSE 0.0.12

LowCodeHub.SSE

Server-Sent Events pub/sub for ASP.NET Core. Publish typed messages to a channel from anywhere; stream them to browsers with one line. Backed by Redis Streams, PostgreSQL, SQL Server, or in-memory, with Last-Event-ID resume, keep-alive pings, auth/ownership hooks, connection limits, health checks and metrics.

dotnet add package LowCodeHub.SSE

Quick start

using LowCodeHub.SSE.Contracts;
using LowCodeHub.SSE.Extensions;

public sealed record OrderEvent(string OrderId, string Status);

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddAuthentication().AddJwtBearer();   // streams require an authenticated user by default
builder.Services.AddAuthorization();
builder.Services.AddSse(builder.Configuration);        // transport chosen from the "Sse" section

var app = builder.Build();
app.UseAuthentication();
app.UseAuthorization();

// Stream: GET /orders/{orderId}/events  (text/event-stream, resumes from Last-Event-ID)
app.MapSse<OrderEvent>("/orders/{orderId}/events", async (http, ct) =>
{
    var orderId = (string)http.GetRouteValue("orderId")!;
    var orders = http.RequestServices.GetRequiredService<IOrderStore>();
    // Ownership check: return null to answer 403.
    return await orders.IsOwnerAsync(orderId, http.User, ct) ? $"orders:{orderId}" : null;
});

// Publish from anywhere
app.MapPost("/orders/{orderId}/ship", async (string orderId, ISsePublisher sse, CancellationToken ct) =>
{
    await sse.PublishAsync($"orders:{orderId}", new OrderEvent(orderId, "Shipped"), ct);
    return Results.Accepted();
});

app.Run();

{
  "Sse": {
    "Provider": "Redis",
    "Redis": { "ConnectionString": "localhost:6379" }
  }
}

const source = new EventSource('/orders/42/events', { withCredentials: true });
source.onmessage = e => console.log(JSON.parse(e.data));          // { orderId: "42", status: "Shipped" }
source.addEventListener('sse-gap', () => reloadOrderFromApi());    // events were trimmed while offline

The browser reconnects automatically and sends Last-Event-ID; the stream replays everything it missed.

Never build a channel from a client-supplied id without an ownership check. /users/{id}/stream that subscribes to user:{id} lets any signed-in user read anyone's feed. Resolve the channel from the authenticated user ($"user:{http.User.FindFirstValue(ClaimTypes.NameIdentifier)}") or verify ownership as above.

Choosing a transport

AddSse(configuration) reads Sse:Provider. When it is absent the provider is inferred from whichever of Sse:Redis, Sse:PostgreSql, Sse:SqlServer has a ConnectionString; with none, in-memory is used. Several connection strings without a Provider is an error.

Code-first registration is also available:

builder.Services.AddSseRedis(o => o.ConnectionString = "localhost:6379");
builder.Services.AddSsePostgreSql(o => o.ConnectionString = cs);
builder.Services.AddSseSqlServer(o => o.ConnectionString = cs);
builder.Services.AddSseInMemory();

// Each has an IConfiguration overload bound from Sse:Redis / Sse:PostgreSql / Sse:SqlServer:
builder.Services.AddSseSqlServer(builder.Configuration);

// Endpoint settings in code (limits, keep-alive, anonymous access):
builder.Services.ConfigureSse(o => o.MaxConnectionsPerUser = 5);

Only one transport is active: the first registration wins (TryAdd).

SQL schema

The SQL transports need their table. The schema script is embedded and exposed by SqlSse.SqlServerResources / SqlSse.PostgreSqlResources (read them with SqlSse.ReadFromResource). With LowCodeHub.Migration.SqlServer / .PostgreSql:

builder.Services.AddSqlServerMigrations(m => m.AddTarget<ISseScriptScanner>(o =>
{
    o.ConnectionString = builder.Configuration.GetConnectionString("Sse")!;
    o.Directories = ["LowCodeHub.SSE.Repositories.SqlServer.Scripts."];
}));

The scripts create [dbo].[SseEvents] / "public"."sse_events". If you change Schema/Table, adapt the scripts. On SQL Server the Channel column uses the binary collation Latin1_General_100_BIN2 (channels are case-sensitive).

Streaming endpoints

// Fixed channel
app.MapSse<MetricSnapshot>("/dashboard/stream", "dashboard:metrics").AllowAnonymous();

// Channel from the request (null => 403)
app.MapSse<Notification>("/me/notifications", http =>
    $"user:{http.User.FindFirstValue(ClaimTypes.NameIdentifier)}");

// Async resolver with ownership checks (null => 403)
app.MapSse<OrderEvent>("/orders/{id}/events", async (http, ct) => ...);

// Inside your own handler (SseResults lives in LowCodeHub.SSE.Http)
app.MapGet("/rooms/{room}/stream", async (string room, ClaimsPrincipal user, IRoomStore rooms) =>
    await rooms.IsMemberAsync(room, user)
        ? SseResults.Stream<ChatMessage>($"room:{room}")
        : Results.Forbid());

MapSse returns a RouteHandlerBuilder, so .RequireAuthorization("policy"), .AllowAnonymous(), .WithTags(...) etc. work as usual.

Every stream served by MapSse / SseResults.Stream does, in order:

1. Authentication: unauthenticated requests get 401 unless the endpoint has AllowAnonymous metadata or Sse:AllowAnonymous is true.
2. Channel authorizers: every registered ISseChannelAuthorizer must allow, else 403.
3. Connection limits: MaxConnections per process and MaxConnectionsPerUser per connection key, else 429 with Retry-After.
4. Resume from the Last-Event-ID header, or ?lastEventId= for clients that cannot set headers.
5. Writes text/event-stream with Cache-Control: no-cache and X-Accel-Buffering: no, and sends a : ping comment every KeepAliveInterval while idle. Comments are invisible to EventSource but keep proxies from closing the connection.

Channel authorizers (tenancy / ownership rules in one place)

public sealed class TenantChannelAuthorizer : ISseChannelAuthorizer
{
    public ValueTask<bool> CanSubscribeAsync(HttpContext http, string channel, CancellationToken ct)
    {
        var tenant = http.User.FindFirstValue("tenant_id");
        return ValueTask.FromResult(tenant is not null && channel.StartsWith($"tenant:{tenant}:", StringComparison.Ordinal));
    }
}

builder.Services.AddSseChannelAuthorizer<TenantChannelAuthorizer>();   // scoped; may inject DbContext

Wire format

id: 1700000000000-0
data: {"orderId":"42","status":"Shipped"}

event: order-cancelled
id: 1700000000001-0
data: {"orderId":"42","status":"Cancelled"}

: ping

event: sse-gap
data: null

The event: line is omitted for the default type message (EventSource.onmessage). Data is serialized with the app's minimal-API JSON options (Microsoft.AspNetCore.Http.Json.JsonOptions, camelCase by default).

Typed event names

await sse.PublishAsync($"orders:{id}", "order-cancelled", evt, ct);   // event: order-cancelled
await sse.PublishAsync($"orders:{id}", evt, ct);                      // default MessageEventType ("message")

source.addEventListener('order-cancelled', e => ...);

Resume gaps

When a client resumes with a Last-Event-ID whose event (and everything older) has been trimmed, the stream first emits event: sse-gap with data: null and no id, then continues with the oldest retained events. Reload state from your API when you see it. Change the name with ResumeGapEventType, or set it to null to disable. No gap event is sent when an ISseHistoryReader<T> is registered, because that reader owns replay.

Publishing

public sealed class OrderNotifier(ISsePublisher sse)
{
    public ValueTask<string> ShippedAsync(Order order, CancellationToken ct) =>
        sse.PublishAsync($"orders:{order.Id}", new OrderEvent(order.Id, "Shipped"), ct);   // returns the event id
}

• Messages are serialized with System.Text.Json web defaults (camelCase). A payload over MaxPayloadBytes (16 KB) throws InvalidOperationException.
• Channels are trimmed of surrounding whitespace and are case-sensitive (tenant:AbC ≠ tenant:abc). They are 1–512 characters; otherwise you get ArgumentException.
• SQL transports and transactions: a publish joins an ambient TransactionScope and becomes visible when it commits. Publishes to the same channel are serialized until commit (sp_getapplock / pg_advisory_xact_lock), so per-channel ids follow commit order and a stream never skips an event that commits late. Keep transactions that publish short.

How delivery works

Each process runs one shared reader per transport. It reads new events for every channel that has live subscribers on that node in one batched query (Redis: one pipelined XREAD per stream, Cluster-safe), then fans them out to in-process subscribers. Any number of clients on a node cost one read, not one per client, and SQL transports never hold a pooled connection per client.

Push (default)

With DeliveryMode = Push, the reader sleeps until something says a channel has new events, then reads just that channel:

The signal only says "channel X has new events". Rows and stream entries stay the source of truth, read from each channel's cursor, so a lost signal cannot lose or reorder events:

• Local publishes wake this process's reader instantly, on every transport. Inside a TransactionScope the wake-up (and the cross-node signal) waits for the commit.
• Safety poll: every SafetyPollInterval (5 s) the reader also reads every active channel, so a missed signal only delays an event. The sse.wakeups.missed counter shows how often that happens.
• Reconnects: if the listener connection drops (failover, pg_terminate_backend, network), it reconnects with backoff (250 ms up to 15 s) and then reads every active channel to catch up. The health check reports Degraded while it is down.
• The listener starts with the first subscriber on a process. Publish-only processes never open one.

SQL Server has no lightweight native push. Service Broker / SqlDependency needs the broker enabled, a queue and per-subscription plumbing, and does not fit a batched multi-channel reader. Without a cross-node signal, publishes on the same process are pushed instantly and publishes on other processes are read every PollInterval (250 ms), the same as Poll mode. For multi-instance deployments, add a Redis wake-up:

{
  "Sse": {
    "SqlServer": {
      "ConnectionString": "Server=...;Database=app;...",
      "RedisWakeup": { "ConnectionString": "localhost:6379" }
    }
  }
}

builder.Services.AddSseSqlServer(o =>
{
    o.ConnectionString = cs;
    o.RedisWakeup = new() { ConnectionMultiplexer = mux };   // or ConnectionString; Channel defaults to "sse:sqlserver:wakeup"
});

// Or bring your own signal (message bus, etc.). It takes precedence over RedisWakeup.
builder.Services.AddSseWakeupSignal<MyBusWakeupSignal>();

ISseWakeupSignal.NotifyAsync(channel) runs after each committed publish. Its failures are logged and counted (sse.notify.errors), and never fail the publish. ListenAsync(sink, ct) calls sink.Connected() once listening and sink.Wakeup(channel) per announcement. It throws or returns when the connection is lost.

Multi-instance notes

• Use the same DeliveryMode on every instance. Only Push publishers send wake-ups.
• Every instance receives every wake-up for the table or prefix, and ignores channels it has no subscribers for. That costs a dictionary lookup.
• PostgreSQL NOTIFY serializes commits of notifying transactions database-wide. At very high publish rates (thousands per second), measure, and consider Poll.

Poll

DeliveryMode = Poll reads every active channel every PollInterval (250 ms). There is no listener and no signal. Local publishes still wake the reader instantly.

Common behaviour

• A resuming client first catches up with direct reads from its Last-Event-ID, then joins the shared feed with no gap or duplicate.
• A slow client that falls SubscriberBufferSize events behind is detached and re-syncs from the transport with its own cursor. Nothing is lost, and other clients are not slowed.
• Read failures are logged and retried with exponential backoff (up to 5 s). Streams stay open. RedisTimeoutException and other TimeoutExceptions count as transient.

Configuration reference

Sse (SseOptions: endpoint behaviour)

Shared transport options (Sse:Redis, Sse:PostgreSql, Sse:SqlServer)

Redis (RedisSseOptions)

An owned multiplexer is created lazily on first use (AbortOnConnectFail=false, timeouts ≥ 15s) and disposed with the container. Push wake-ups use the pub/sub channel {StreamKeyPrefix}wakeup. Pub/sub channels do not share the key space, so this channel cannot collide with a stream.

PostgreSQL / SQL Server (PostgreSqlSseOptions / SqlServerSseOptions)

All options are validated at startup (ValidateOnStart). Invalid values fail host start with OptionsValidationException.

Advanced

Consuming streams without HTTP helpers

ISseStreamService.StreamAsync<T>(channel, lastEventId, ct) returns IAsyncEnumerable<SseItem<T>>. You can pass it to TypedResults.ServerSentEvents(...) or iterate it yourself. It does not apply authentication, authorizers, limits or : ping comments, because those live in MapSse / SseResults. Without : ping comments, register an ISseHeartbeatFactory<T> so idle connections survive proxies. A gap item has EventType == "sse-gap" and Data == default.

History readers (replay from your own store)

public sealed class OrderHistory(AppDbContext db) : ISseHistoryReader<OrderEvent>
{
    public async IAsyncEnumerable<SseHistoryEntry<OrderEvent>> ReadAfterAsync(
        string channel, string? lastEventId, [EnumeratorCancellation] CancellationToken ct = default)
    {
        // Yield entries after lastEventId; use transport ids as EventId so the live tail continues exactly after them.
        ...
    }
}

builder.Services.AddSseHistoryReader<OrderEvent, OrderHistory>();   // scoped: a DbContext is fine

The reader runs (in its own DI scope) before the live tail. If the last replayed EventId is a valid transport id, the live tail resumes right after it with no duplicates. Otherwise it resumes from lastEventId, or from "now".

Typed heartbeats

builder.Services.AddSseHeartbeatFactory<OrderEvent, OrderHeartbeat>();   // emits event: heartbeat with your payload

These are only needed when clients want a typed heartbeat. MapSse already keeps connections alive with comments.

AddSseRedis<TMessage, THistory, THeartbeat>(...) (and the PostgreSQL / SQL Server equivalents) register both in one call.

Redis with startup fallback

bool usedRedis = services.AddSseWithMemoryFallback(o => ..., probeTimeoutMs: 800); probes Redis once at registration and falls back to in-memory for the life of the process. Log the result.

Health checks

Durable transports register a readiness check, sse-redis / sse-postgresql / sse-sqlserver, tagged sse and readiness. It runs a Redis PING, or a zero-row query against the events table (which also catches a missing schema). It reports Degraded when the push listener has been disconnected for more than 2 seconds. Events still flow through the safety poll in that state, only slower. Map it with app.MapHealthChecks("/health/ready", new() { Predicate = r => r.Tags.Contains("readiness") }).

Observability

builder.Services.AddOpenTelemetry()
    .WithMetrics(m => m.AddMeter(SseDiagnostics.MeterName))
    .WithTracing(t => t.AddSource(SseDiagnostics.ActivitySourceName));

Publishing creates an sse.publish producer activity.

Requirements

• .NET 10, ASP.NET Core
• Redis 5+ (Streams), PostgreSQL 12+, or SQL Server 2016+
• PostgreSQL push: the listener needs a session-level connection. Behind PgBouncer in transaction mode LISTEN does not work, so point the connection string at PostgreSQL directly (or a session-mode pool), or use DeliveryMode = Poll.

License

MIT