Client connections

Most realtime needs don't need a connection. A deferred response to one request is request-driven streaming; server→client facts are client events; and producer-owned ordered outputs are streams. A connection is a stateful, bidirectional carrier for the cases those deliberately don't cover — when the conversation itself is stateful or latency-interactive. It does not add replay, ordering, or delivery guarantees to the contract it carries:

1. Interactive rate — client→server messages are frequent and per-message latency is part of the UX (collaboration ops, live input, telemetry ingest where batching's flush window is itself too slow).
2. The connection is the state — device links where connect/disconnect are business events, or the link fronts a stateful device protocol (the IoT gateway shape).
3. Server→client RPC — the server must address one specific connected client and await a result.

If a list-carrying batch command flushed every N items or M milliseconds would do, use that instead: one authorization, one validation pass, one transaction per batch, visible in the schema.

In this section

The kernel

Elarion.Connections (contracts in Elarion.Abstractions.Connections) is transport-neutral — the same seams serve a WebSocket adapter today and a raw TCP or datagram adapter tomorrow, and nothing in them assumes a socket exists at all:

Register the kernel once:

builder.Services.AddElarionConnections();   // registry + client-events bridge

Identity promotion (anonymous → authenticated)

A connection may register anonymously — the shape of device protocols whose authentication is a framed exchange after connect rather than a connect-time credential — and later perform exactly one atomic, one-way promotion to an authenticated identity:

var status = await registry.PromoteAsync(connection.Connection.ConnectionId,
    new ClientConnectionIdentity {
        Principal = principal,
        PrincipalId = deviceId,
        Metadata = new Dictionary<string, string> { ["firmware"] = version },
    }, ct);
// Promoted | AlreadyAuthenticated | ConnectionNotFound — demotion and user switching are rejected.

The rules are deliberate and security-shaped:

• Stable facts never change (ConnectionId, Transport, ConnectedAt); the principal, principal id, metadata, and revision are replaced together as one immutable snapshot. Inputs are cloned and metadata is defensively copied and bounded — mutating what you passed in cannot alter stored identity.
• Only anonymous/no-id → authenticated/non-empty-id is accepted, exactly once — concurrent promoters get one winner; demotion and principal switching are rejected as AlreadyAuthenticated.
• An in-flight dispatch keeps the snapshot it captured at its boundary; the next dispatch observes the promoted snapshot. A failed validation or a lost race leaves the prior snapshot and revision intact; a promotion observer failure is isolated and never rolls the identity back.
• Client-event subscriptions are disposed on promotion — the peer must resubscribe so subscribe-time authorization is reevaluated under the new identity. Observers implementing OnIdentityPromotedAsync(previous, current, …) see both snapshots; disconnect observers and the codec's OnClosedAsync always receive the final (latest) snapshot.

Talking back to one client

Look a connection up and use its sink — from a handler, an observer, or an actor turn:

// All channels of one device (or all tabs of one user):
foreach (var sink in registry.GetForPrincipal(deviceId)) {
    await ((WebSocketClientConnection)sink).SendTextAsync(frame, ct);   // raw leg, codec-owned format
}

// Request/reply into a client whose codec implements InvokeAsync:
var status = await sink.InvokeAsync<StatusQuery, StatusReply>(
    "status.get", new StatusQuery(), new ClientInvokeOptions { Timeout = TimeSpan.FromSeconds(5) }, ct);

InvokeAsync is deliberately the simple tier — one request, one reply, bounded by a timeout. Multi-message protocol conversations (sequence numbers, re-synchronisation, wait-for-any flows) are codec/actor state; the codec seam is their mounting point, not their replacement — see device gateways.

Every invoke is bounded by default: a call that carries no ClientInvokeOptions.Timeout gets the kernel's DefaultInvokeTimeout (30 s out of the box; AddElarionConnections(o => o.DefaultInvokeTimeout = …)), resolved by the sink before the codec sees the options — a client that never answers surfaces as a TimeoutException, never a silently hung await. The layering is per-call Timeout > DefaultInvokeTimeout > unbounded: a per-call value always wins (pass Timeout.InfiniteTimeSpan to make a single call unbounded), and only a default explicitly configured to null applies none.

Client events over a connection

A connection can serve the same topics the SSE endpoint serves — same catalog, same fail-closed authorization, so a second transport can never fork the rules. The adapter parses its subscribe frame and hands the requests to the bridge; delivery framing is the adapter's choice:

var result = await bridge.SubscribeAsync(connection.Connection, requests,
    (envelope, ct) => connection.SendTextAsync($"event:{envelope.Topic}:{envelope.Payload}", ct), ct);

Delivery starts with the elarion.connected greeting (the same re-query contract an SSE stream opens with), and the subscription dies automatically when its connection unregisters. Statuses mirror the SSE endpoint's HTTP mapping (unauthenticated / invalid / not-found — unknown and denied topics stay indistinguishable).

Rules of the road

• Facts still travel as client events, even over a connection — the sink is for conversation-shaped traffic (replies, control frames, RPC), not a side channel for state fan-out.
• Guarantees are unchanged: events are at-most-once hints healed by re-query; sends are at-most-once; InvokeAsync faults explicitly. Commands a device may retransmit belong behind [Idempotent].
• One device, many connections: give every channel the same PrincipalId and let a digital-twin actor keyed by device id serialize the shared state — per-connection ordering comes from the receive loop, cross-channel consistency from the actor.
• The registry is node-local by design. Multi-node deployments co-locate device ingress with the single-homed twin (the role-holder proxy's prefix list); a replicated connection directory is the point where you adopt a clustered runtime instead.
• Telemetry: register the Elarion.Connections meter — connection.active, connection.opened/connection.closed (tagged elarion.connection.transport), connection.identity_promotions (tagged elarion.connection.promotion.outcome), and connection.event_subscriptions.active — and, for TCP endpoints, Elarion.Connections.Tcp: elarion.tcp.tls.handshake.duration (seconds, tagged elarion.tcp.tls.outcome), elarion.tcp.connection.failures (bounded elarion.tcp.failure.stage), elarion.tcp.connection.closed (elarion.tcp.close.mode = graceful/forced), elarion.tcp.outbound.pending/elarion.tcp.outbound.saturated, and elarion.tcp.idle. Tags are fixed vocabularies only — never connection/principal ids, payloads, endpoints, operation names, or certificate data.