SignalR

<Project Sdk="Microsoft.NET.Sdk.Web">
  <PropertyGroup>
    <TargetFramework>net10.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="Microsoft.AspNetCore.SignalR.StackExchangeRedis" Version="10.0.13" />
  </ItemGroup>
</Project>

using Chatterbox;
using StackExchange.Redis;

var builder = WebApplication.CreateBuilder(args);
var redisConnection = builder.Configuration["REDIS"] ?? "localhost:6379";

builder.Services.AddSingleton<IConnectionMultiplexer>(_ => ConnectionMultiplexer.Connect(redisConnection));
builder.Services.AddSingleton<ChatStore>();
builder.Services.AddSignalR(o =>
{
    o.KeepAliveInterval = TimeSpan.FromSeconds(20);      // heartbeatMs
    o.ClientTimeoutInterval = TimeSpan.FromSeconds(40);  // 2 x heartbeatMs
    o.MaximumReceiveMessageSize = 16384;                 // maxFrameBytes
    o.MaximumParallelInvocationsPerClient = 1;           // frames of one connection run in order
})
.AddStackExchangeRedis(redisConnection, o =>
    o.Configuration.ChannelPrefix = RedisChannel.Literal("chatterbox"));

var app = builder.Build();

// Stand-in for a real login: returns a one-time ticket valid for 30 seconds.
app.MapPost("/api/ticket", async (string user, ChatStore store) =>
    Results.Ok(new { ticket = await store.IssueTicketAsync(user) }));

app.MapGet("/healthz", () => Results.Ok(new { instance = Environment.MachineName }));
app.MapHub<ChatHub>("/hubs/chatterbox");
app.Run();

using System.Security.Cryptography;
using System.Text.Json.Nodes;
using StackExchange.Redis;

namespace Chatterbox;

public sealed class ChatException(string code, string message, bool retryable = false) : Exception(message)
{
    public string Code { get; } = code;
    public bool Retryable { get; } = retryable;
}

/// <summary>20 tokens per second, capacity 40, per connection.</summary>
public sealed class TokenBucket
{
    private double _tokens = 40;
    private long _last = Environment.TickCount64;
    public int Violations;

    public bool TryTake()
    {
        var now = Environment.TickCount64;
        _tokens = Math.Min(40, _tokens + (now - _last) / 1000.0 * 20);
        _last = now;
        if (_tokens < 1) return false;
        _tokens -= 1;
        return true;
    }
}

public static class Ids
{
    private const string Alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";

    /// <summary>26-character ULID: 48-bit time plus 80 random bits, Crockford base32.</summary>
    public static string Ulid()
    {
        Span<byte> b = stackalloc byte[16];
        var ms = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
        for (var i = 5; i >= 0; i--) { b[i] = (byte)(ms & 0xFF); ms >>= 8; }
        RandomNumberGenerator.Fill(b[6..]);
        var chars = new char[26];
        var bits = new System.Numerics.BigInteger(b, isUnsigned: true, isBigEndian: true);
        for (var i = 25; i >= 0; i--) { chars[i] = Alphabet[(int)(bits & 31)]; bits >>= 5; }
        return new string(chars);
    }
    public static string Server() => "s_" + Ulid();
}

public sealed class ChatStore(IConnectionMultiplexer redis)
{
    private IDatabase Db => redis.GetDatabase();

    public async Task<string> IssueTicketAsync(string user)
    {
        var ticket = Convert.ToHexString(RandomNumberGenerator.GetBytes(16)).ToLowerInvariant();
        await Db.StringSetAsync($"ticket:{ticket}", user, TimeSpan.FromSeconds(30));
        return ticket;
    }

    // GETDEL makes the ticket one-time even across several server instances.
    public async Task<string?> RedeemTicketAsync(string ticket)
        => (string?)await Db.StringGetDeleteAsync($"ticket:{ticket}");

    public Task<long> NextSeqAsync(string room) => Db.StringIncrementAsync($"room:{room}:seq");

    public async Task<long> LastSeqAsync(string room)
        => (long)await Db.StringGetAsync($"room:{room}:seq");

    public Task AddMessageAsync(string room, long seq, JsonNode envelope)
        => Db.SortedSetAddAsync($"room:{room}:messages", envelope.ToJsonString(), seq);

    public async Task<List<JsonNode>> MessagesAsync(string room, long? before, int limit)
    {
        var max = before is null ? double.PositiveInfinity : before.Value;
        var exclude = before is null ? Exclude.None : Exclude.Stop;
        var values = await Db.SortedSetRangeByScoreAsync($"room:{room}:messages", double.NegativeInfinity, max,
            exclude, Order.Descending, 0, limit);
        return values.Select(v => JsonNode.Parse((string)v!)!).Reverse().ToList();
    }

    public async Task<List<JsonNode>> SinceAsync(string room, long since)
    {
        var values = await Db.SortedSetRangeByScoreAsync($"room:{room}:messages", since, double.PositiveInfinity,
            Exclude.Start, Order.Ascending);
        return values.Select(v => JsonNode.Parse((string)v!)!).ToList();
    }

    public async Task<string?> GetAckAsync(string user, string id)
        => await Db.StringGetAsync($"ack:{user}:{id}");

    public Task SetAckAsync(string user, string id, string ack)
        => Db.StringSetAsync($"ack:{user}:{id}", ack, TimeSpan.FromHours(24));

    public Task<bool> AddMemberAsync(string room, string user)
        => Db.SetAddAsync($"room:{room}:members", user);
    public Task<bool> RemoveMemberAsync(string room, string user)
        => Db.SetRemoveAsync($"room:{room}:members", user);
    public Task<bool> IsMemberAsync(string room, string user)
        => Db.SetContainsAsync($"room:{room}:members", user);
    public async Task<string[]> MembersAsync(string room)
        => (await Db.SetMembersAsync($"room:{room}:members")).Select(v => (string)v!).ToArray();
    public async Task<string[]> RoomsOfAsync(string user)
        => (await Db.SetMembersAsync($"user:{user}:rooms")).Select(v => (string)v!).ToArray();
    public Task TrackRoomAsync(string user, string room, bool add)
        => add ? Db.SetAddAsync($"user:{user}:rooms", room) : Db.SetRemoveAsync($"user:{user}:rooms", room);

    public Task<long> ConnectionsAsync(string user, int delta)
        => Db.StringIncrementAsync($"user:{user}:conns", delta);
    public async Task<long> ConnectionCountAsync(string user)
        => (long)await Db.StringGetAsync($"user:{user}:conns");

    public Task SetStatusAsync(string user, string status) => Db.StringSetAsync($"user:{user}:status", status);
    public async Task<string> StatusAsync(string user)
        => (string?)await Db.StringGetAsync($"user:{user}:status") ?? "online";
}

using System.Runtime.CompilerServices;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.RegularExpressions;
using Microsoft.AspNetCore.SignalR;

namespace Chatterbox;

public sealed partial class ChatHub(ChatStore store, IHubContext<ChatHub> hubs, IConfiguration config) : Hub
{
    [GeneratedRegex("^[a-z0-9-]{1,64}$")] private static partial Regex RoomName();

    private static readonly string[] AckedTypes = ["join", "leave", "message", "presence", "history"];
    private string? UserId => Context.Items["uid"] as string;

    // ---------- connection lifecycle ----------

    public override async Task OnConnectedAsync()
    {
        Context.Items["bucket"] = new TokenBucket();
        var connectionId = Context.ConnectionId;
        var context = Context;
        // hello must arrive within 5 s, otherwise the server aborts the connection
        _ = Task.Run(async () =>
        {
            await Task.Delay(5000);
            if (context.Items["uid"] is null)
            {
                await hubs.Clients.Client(connectionId).SendAsync("frame",
                    Envelope("error", null, ErrorPayload(null, "unauthorized", "hello timeout", false)));
                context.Abort();
            }
        });
        await base.OnConnectedAsync();
    }

    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        if (UserId is { } user)
        {
            await store.ConnectionsAsync(user, -1);
            _ = Task.Run(() => OfflineAfterGraceAsync(user));
        }
        await base.OnDisconnectedAsync(exception);
    }

    private async Task OfflineAfterGraceAsync(string user)
    {
        await Task.Delay(config.GetValue("PRESENCE_GRACE_MS", 15000));
        if (await store.ConnectionCountAsync(user) > 0) return;
        foreach (var room in await store.RoomsOfAsync(user))
            await hubs.Clients.Group(Group(room)).SendAsync("frame",
                Envelope("presence", room, new JsonObject { ["userId"] = user, ["name"] = user, ["status"] = "offline" }));
    }

    // ---------- the single entry point: one Chatterbox envelope per invocation ----------

    public async Task Frame(JsonElement raw)
    {
        var frame = JsonNode.Parse(raw.GetRawText())!.AsObject();
        var id = frame["id"]?.GetValue<string>();
        try
        {
            if (!((TokenBucket)Context.Items["bucket"]!).TryTake())
            {
                var bucket = (TokenBucket)Context.Items["bucket"]!;
                if (++bucket.Violations >= 10) { await Fail(id, "rate_limited", "slow down", true); Context.Abort(); return; }
                throw new ChatException("rate_limited", "slow down", true);
            }
            if (frame["v"]?.GetValue<int>() != 1)
            {
                await Fail(id, "unsupported_version", "only v1", false);
                Context.Abort();
                return;
            }
            var type = frame["type"]?.GetValue<string>() ?? throw new ChatException("bad_request", "type missing");
            if (id is null || id.Length is < 26 or > 36) throw new ChatException("bad_request", "id must be 26-36 chars");
            var payload = frame["payload"] as JsonObject ?? throw new ChatException("bad_request", "payload must be an object");
            var room = frame["room"]?.GetValue<string>();

            if (type == "hello") { await Hello(id, payload); return; }
            if (UserId is not { } user)
            {
                await Fail(id, "unauthorized", "send hello first", false);
                Context.Abort();
                return;
            }
            if (room is not null && !RoomName().IsMatch(room)) throw new ChatException("bad_request", "bad room name");

            if (AckedTypes.Contains(type) && await store.GetAckAsync(user, id) is { } cached)
            {
                var again = JsonNode.Parse(cached)!.AsObject();
                again["payload"]!["dup"] = true;       // same ack, no new seq, no re-broadcast
                again["ts"] = Now();
                await Clients.Caller.SendAsync("frame", again);
                return;
            }

            var (seq, result) = type switch
            {
                "join" => await Join(user, Need(room), payload),
                "leave" => await Leave(user, Need(room)),
                "message" => await Message(user, id, Need(room), payload),
                "presence" => await Presence(user, payload),
                "history" => await History(Need(room), payload),
                "typing" => await Typing(user, Need(room), payload),
                _ => throw new ChatException("bad_request", $"type '{type}' is not implemented here"),
            };
            if (type == "typing") return;               // fire-and-forget: no ack

            var ack = Envelope("ack", room, new JsonObject
            {
                ["ref"] = id, ["ok"] = true, ["seq"] = seq, ["dup"] = false, ["result"] = result,
            });
            await store.SetAckAsync(user, id, ack.ToJsonString());   // only acks are remembered
            await Clients.Caller.SendAsync("frame", ack);
        }
        catch (ChatException e)
        {
            await Fail(id, e.Code, e.Message, e.Retryable);
        }
    }

    // Server-to-client stream: the whole room history, one message envelope per stream item.
    public async IAsyncEnumerable<JsonNode> ExportHistory(string room, [EnumeratorCancellation] CancellationToken ct)
    {
        var user = UserId ?? throw new HubException("send hello first");
        if (!RoomName().IsMatch(room) || !await store.IsMemberAsync(room, user)) throw new HubException("forbidden");
        long? before = null;
        while (true)
        {
            var page = await store.MessagesAsync(room, before, 50);
            if (page.Count == 0) yield break;
            foreach (var m in Enumerable.Reverse(page)) { ct.ThrowIfCancellationRequested(); yield return m; }
            before = page[0]["payload"]!["seq"]!.GetValue<long>();
        }
    }

    // ---------- handlers ----------

    private async Task Hello(string id, JsonObject payload)
    {
        if (UserId is not null) throw new ChatException("conflict", "already authenticated");
        var ticket = payload["token"]?.GetValue<string>();
        var user = ticket is null ? null : await store.RedeemTicketAsync(ticket);
        if (user is null)
        {
            await Fail(id, "unauthorized", "invalid or used ticket", false);
            Context.Abort();
            return;
        }
        Context.Items["uid"] = user;
        await store.ConnectionsAsync(user, 1);
        await Clients.Caller.SendAsync("frame", Envelope("hello", null, new JsonObject
        {
            ["ref"] = id, ["userId"] = user, ["name"] = user, ["sessionId"] = Ids.Ulid(),
            ["heartbeatMs"] = 20000, ["maxFrameBytes"] = 16384, ["resumed"] = false,
            ["rooms"] = new JsonArray((await store.RoomsOfAsync(user)).Select(r => (JsonNode)r).ToArray()),
        }));
    }

    private async Task<(long?, JsonObject)> Join(string user, string room, JsonObject payload)
    {
        if (room.StartsWith("dm-") && !room.Split('-').Contains(user))
            throw new ChatException("forbidden", "not your direct-message room");
        var added = await store.AddMemberAsync(room, user);
        await store.TrackRoomAsync(user, room, true);
        await Groups.AddToGroupAsync(Context.ConnectionId, Group(room));
        if (added)
            await Clients.GroupExcept(Group(room), Context.ConnectionId).SendAsync("frame", Envelope("presence", room,
                new JsonObject { ["userId"] = user, ["name"] = user, ["status"] = await store.StatusAsync(user) }));
        var members = new JsonArray();
        foreach (var m in await store.MembersAsync(room))
            members.Add(new JsonObject { ["userId"] = m, ["name"] = m, ["status"] = await store.StatusAsync(m) });
        if (payload["since"] is JsonValue s && s.TryGetValue<long>(out var since))
            foreach (var missed in await store.SinceAsync(room, since))
                await Clients.Caller.SendAsync("frame", missed);
        return (null, new JsonObject { ["members"] = members, ["lastSeq"] = await store.LastSeqAsync(room) });
    }

    private async Task<(long?, JsonObject)> Leave(string user, string room)
    {
        if (!await store.RemoveMemberAsync(room, user)) throw new ChatException("not_found", "not a member");
        await store.TrackRoomAsync(user, room, false);
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, Group(room));
        return (null, new JsonObject());
    }

    private async Task<(long?, JsonObject)> Message(string user, string id, string room, JsonObject payload)
    {
        if (!await store.IsMemberAsync(room, user)) throw new ChatException("forbidden", "join the room first");
        var text = payload["text"]?.GetValue<string>();
        if (string.IsNullOrEmpty(text) || text.Length > 4000) throw new ChatException("bad_request", "text must be 1..4000");
        var seq = await store.NextSeqAsync(room);
        var broadcast = new JsonObject
        {
            ["v"] = 1, ["type"] = "message", ["id"] = id, ["room"] = room, ["ts"] = Now(),
            ["payload"] = new JsonObject
            {
                ["from"] = new JsonObject { ["userId"] = user, ["name"] = user },
                ["text"] = text, ["replyTo"] = payload["replyTo"]?.DeepClone(),
                ["attachments"] = payload["attachments"]?.DeepClone() ?? new JsonArray(),
                ["seq"] = seq, ["edited"] = false,
            },
        };
        await store.AddMessageAsync(room, seq, broadcast);
        await Clients.Group(Group(room)).SendAsync("frame", broadcast);   // includes the sender
        return (seq, new JsonObject());
    }

    private async Task<(long?, JsonObject)> Presence(string user, JsonObject payload)
    {
        var status = payload["status"]?.GetValue<string>();
        if (status is not ("online" or "away")) throw new ChatException("bad_request", "status");
        var changed = await store.StatusAsync(user) != status;
        await store.SetStatusAsync(user, status);
        if (changed)
            foreach (var room in await store.RoomsOfAsync(user))
                await Clients.Group(Group(room)).SendAsync("frame", Envelope("presence", room,
                    new JsonObject { ["userId"] = user, ["name"] = user, ["status"] = status }));
        return (null, new JsonObject());
    }

    private async Task<(long?, JsonObject)> History(string room, JsonObject payload)
    {
        var user = UserId!;
        if (!await store.IsMemberAsync(room, user)) throw new ChatException("forbidden", "join the room first");
        var limit = Math.Clamp(payload["limit"]?.GetValue<int>() ?? 50, 1, 100);
        long? before = payload["before"] is JsonValue b && b.TryGetValue<long>(out var bv) ? bv : null;
        var page = await store.MessagesAsync(room, before, limit + 1);   // one extra row tells hasMore
        var hasMore = page.Count > limit;
        if (hasMore) page.RemoveAt(0);
        long? next = hasMore ? page[0]["payload"]!["seq"]!.GetValue<long>() : null;
        return (null, new JsonObject { ["messages"] = new JsonArray(page.ToArray()), ["hasMore"] = hasMore, ["next"] = next });
    }

    private async Task<(long?, JsonObject)> Typing(string user, string room, JsonObject payload)
    {
        if (!await store.IsMemberAsync(room, user)) throw new ChatException("forbidden", "join the room first");
        await Clients.GroupExcept(Group(room), Context.ConnectionId).SendAsync("frame", Envelope("typing", room,
            new JsonObject { ["userId"] = user, ["state"] = payload["state"]?.GetValue<string>() ?? "start" }));
        return (null, new JsonObject());
    }

    // ---------- helpers ----------

    private static string Group(string room) => "room:" + room;
    private static long Now() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
    private static string Need(string? room) => room ?? throw new ChatException("bad_request", "room required");

    private static JsonObject Envelope(string type, string? room, JsonObject payload) => new()
    {
        ["v"] = 1, ["type"] = type, ["id"] = Ids.Server(), ["room"] = room, ["ts"] = Now(), ["payload"] = payload,
    };

    private static JsonObject ErrorPayload(string? @ref, string code, string message, bool retryable) => new()
    {
        ["ref"] = @ref, ["code"] = code, ["message"] = message, ["retryable"] = retryable,
    };

    private Task Fail(string? @ref, string code, string message, bool retryable)
        => Clients.Caller.SendAsync("frame", Envelope("error", null, ErrorPayload(@ref, code, message, retryable)));
}

FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
WORKDIR /src
COPY Chatterbox.csproj .
RUN dotnet restore
COPY . .
RUN dotnet publish -c Release -o /app --no-restore

FROM mcr.microsoft.com/dotnet/aspnet:10.0
WORKDIR /app
COPY --from=build /app .
ENV ASPNETCORE_URLS=http://+:8080
ENTRYPOINT ["dotnet", "Chatterbox.dll"]

services:
  redis:
    image: redis:8.8.3-alpine
    ports:
      - "6390:6379"
  chat-a:
    build: .
    environment:
      REDIS: redis:6379
      PRESENCE_GRACE_MS: "2000"
    ports:
      - "8190:8080"
    depends_on: [redis]
  chat-b:
    build: .
    environment:
      REDIS: redis:6379
      PRESENCE_GRACE_MS: "2000"
    ports:
      - "8191:8080"
    depends_on: [redis]

cd chatterbox-signalr/server
docker compose -p chatterbox-signalr up -d --build      # redis on 6390, instances on 8190 and 8191
cd ../client
npm install
node check.mjs                                          # two clients, two servers, one Redis
docker compose -p chatterbox-signalr down -v

import { HubConnectionBuilder, HttpTransportType, LogLevel } from "@microsoft/signalr";

const ulid = () => crypto.randomUUID().replaceAll("-", "").slice(0, 26).toUpperCase();

export class ChatterboxClient extends EventTarget {
  #conn; #pending = new Map(); #rooms = new Map(); // room -> last seq seen
  #base; #user;

  constructor(baseUrl, user) {
    super();
    this.#base = baseUrl;
    this.#user = user;
    this.#conn = new HubConnectionBuilder()
      .withUrl(`${baseUrl}/hubs/chatterbox`, { transport: HttpTransportType.WebSockets })
      .withAutomaticReconnect([0, 1000, 3000, 10000])
      .withKeepAliveInterval(20_000)   // matches the server's KeepAliveInterval (heartbeatMs)
      .withServerTimeout(40_000)       // 2 x the keep-alive interval
      .configureLogging(LogLevel.Warning)
      .build();

    this.#conn.on("frame", (env) => this.#onFrame(env));
    this.#conn.onreconnected(async () => {
      await this.#hello();                                  // a new connection id needs a new hello
      for (const [room, seq] of this.#rooms) await this.join(room, seq || null);  // groups are not persisted
      this.dispatchEvent(new Event("reconnected"));
    });
  }

  async start() {
    await this.#conn.start();
    return this.#hello();
  }

  async #hello() {
    const res = await fetch(`${this.#base}/api/ticket?user=${this.#user}`, { method: "POST" });
    const { ticket } = await res.json();
    return this.#send("hello", null, { token: ticket, client: "web/1.4" });
  }

  #send(type, room, payload) {
    const env = { v: 1, type, id: ulid(), room, ts: 0, payload };
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.#pending.delete(env.id);
        reject(new Error(`ack timeout for ${type}`));
      }, 10_000);
      this.#pending.set(env.id, { resolve, reject, timer });
      this.#conn.send("Frame", env).catch(reject);          // fire-and-forget; the ack arrives as a frame
    });
  }

  #onFrame(env) {
    const ref = env.payload?.ref;
    const waiter = ref && this.#pending.get(ref);
    if (waiter && ["ack", "error", "hello"].includes(env.type)) {
      clearTimeout(waiter.timer);
      this.#pending.delete(ref);
      if (env.type === "error") waiter.reject(Object.assign(new Error(env.payload.message), env.payload));
      else waiter.resolve(env.payload);
      return;
    }
    const seq = env.payload?.seq;
    if (seq && env.room) this.#rooms.set(env.room, Math.max(this.#rooms.get(env.room) ?? 0, seq));
    this.dispatchEvent(new CustomEvent(env.type, { detail: env }));
  }

  async join(room, since = null) {
    if (!this.#rooms.has(room)) this.#rooms.set(room, 0);
    return this.#send("join", room, { since });
  }
  message(room, text) { return this.#send("message", room, { text, replyTo: null, attachments: [] }); }
  history(room, before = null, limit = 50) { return this.#send("history", room, { before, limit }); }
  presence(status) { return this.#send("presence", null, { status }); }
  typing(room, state) { this.#conn.send("Frame", { v: 1, type: "typing", id: ulid(), room, ts: 0, payload: { state } }); }
  // resend an envelope with its ORIGINAL id (idempotent retry)
  raw(env) { return new Promise((resolve) => { this.#pending.set(env.id, { resolve, reject: resolve, timer: setTimeout(() => {}, 0) }); this.#conn.send("Frame", env); }); }
  exportHistory(room, onItem) {
    return new Promise((resolve, reject) =>
      this.#conn.stream("ExportHistory", room).subscribe({ next: onItem, error: reject, complete: resolve }));
  }
  stop() { return this.#conn.stop(); }
}

import assert from "node:assert/strict";
import { ChatterboxClient } from "./chatterbox-client.mjs";

const room = "room-" + Math.random().toString(36).slice(2, 8);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const alice = new ChatterboxClient("http://localhost:8190", "alice");   // instance A
const bob = new ChatterboxClient("http://localhost:8191", "bob");       // instance B
await alice.start();
await bob.start();

const bobMsgs = [];
bob.addEventListener("message", (e) => bobMsgs.push(e.detail));
await alice.join(room);
const joined = await bob.join(room);
assert.deepEqual(joined.result.members.map((m) => m.userId).sort(), ["alice", "bob"]);

const a1 = await alice.message(room, "hello from A");
const a2 = await alice.message(room, "second");
assert.deepEqual([a1.seq, a2.seq], [1, 2]);          // sequence numbers come from Redis INCR
await sleep(300);
assert.deepEqual(bobMsgs.map((e) => e.payload.seq), [1, 2]);   // delivered via the backplane

const env = { v: 1, type: "message", id: "01JZ" + Math.random().toString(36).slice(2).toUpperCase().padEnd(22, "A").slice(0, 22), room, ts: 0,
              payload: { text: "once", replyTo: null, attachments: [] } };
const first = await alice.raw(env);
const second = await alice.raw(env);                 // same id again
assert.equal(first.dup, false);
assert.equal(second.dup, true);
assert.equal(first.seq, second.seq);

await assert.rejects(alice.message("nowhere", "x"), (e) => e.code === "forbidden");
await assert.rejects(alice.message(room, ""), (e) => e.code === "bad_request");
await alice.stop();
await bob.stop();
console.log("OK");

using System.Diagnostics;
using System.Text;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Connections;
using Microsoft.AspNetCore.SignalR;
using Microsoft.IdentityModel.Tokens;

var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
    .AddJwtBearer(o =>
    {
        o.TokenValidationParameters = new TokenValidationParameters
        {
            ValidIssuer = "https://chat.example.test",
            ValidAudience = "chatterbox",
            IssuerSigningKey = new SymmetricSecurityKey(Encoding.UTF8.GetBytes("change-me-change-me-change-me-32b")),
        };
        o.Events = new JwtBearerEvents
        {
            OnMessageReceived = context =>
            {
                var token = context.Request.Query["access_token"];
                if (!string.IsNullOrEmpty(token) && context.HttpContext.Request.Path.StartsWithSegments("/hubs"))
                    context.Token = token;
                return Task.CompletedTask;
            },
        };
    });
builder.Services.AddAuthorization();
builder.Services.AddCors(o => o.AddPolicy("chat", p => p
    .WithOrigins("https://chat.example.test").WithMethods("GET", "POST", "DELETE").AllowAnyHeader().AllowCredentials()));

builder.Services.AddSingleton<AuditFilter>();
builder.Services
    .AddSignalR(o => o.AddFilter<AuditFilter>())
    .AddMessagePackProtocol()
    .AddAzureSignalR(o => o.ConnectionString = "Endpoint=https://example.service.signalr.net;AccessKey=change-me;Version=1.0;");

var app = builder.Build();
app.UseWebSockets(new Microsoft.AspNetCore.Builder.WebSocketOptions { AllowedOrigins = { "https://chat.example.test" } });
app.UseCors();
app.UseAuthentication();
app.UseAuthorization();
app.MapHub<ChatHub>("/hubs/chatterbox", o =>
{
    o.Transports = HttpTransportType.WebSockets;
    o.AllowStatefulReconnects = true;
    o.CloseOnAuthenticationExpiration = true;
}).RequireCors("chat");
app.Run();

public sealed class AuditFilter(ILogger<AuditFilter> log) : IHubFilter
{
    public async ValueTask<object?> InvokeMethodAsync(HubInvocationContext ctx, Func<HubInvocationContext, ValueTask<object?>> next)
    {
        var watch = Stopwatch.StartNew();
        try { return await next(ctx); }
        finally { log.LogInformation("{Method} took {Ms} ms", ctx.HubMethodName, watch.ElapsedMilliseconds); }
    }
}

[Microsoft.AspNetCore.Authorization.Authorize]
public sealed class ChatHub : Hub { }