Atmosphere

package example.chatterbox;

import org.atmosphere.cpr.ApplicationConfig;
import org.atmosphere.cpr.AtmosphereServlet;
import org.eclipse.jetty.ee10.servlet.ServletContextHandler;
import org.eclipse.jetty.ee10.servlet.ServletHolder;
import org.eclipse.jetty.ee10.websocket.jakarta.server.config.JakartaWebSocketServletContainerInitializer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;

public final class Main {
    public static void main(String[] args) throws Exception {
        int port = Integer.parseInt(System.getenv().getOrDefault("PORT", "8080"));
        Server server = new Server();
        ServerConnector connector = new ServerConnector(server);
        connector.setPort(port);
        server.addConnector(connector);

        ServletContextHandler context = new ServletContextHandler("/");
        // The Jakarta WebSocket container must exist before AtmosphereServlet initialises.
        JakartaWebSocketServletContainerInitializer.configure(context,
                (servletContext, container) -> container.addEndpoint(PlainEcho.class));

        ServletHolder atmosphere = new ServletHolder(AtmosphereServlet.class);
        atmosphere.setInitParameter(ApplicationConfig.ANNOTATION_PACKAGE, "example.chatterbox");
        atmosphere.setInitParameter(ApplicationConfig.WEBSOCKET_SUPPORT, "true");
        atmosphere.setInitParameter(ApplicationConfig.WEBSOCKET_MAXTEXTSIZE, "16384");
        // Replace Atmosphere's built-in same-origin check with an explicit allow-list.
        atmosphere.setInitParameter(ApplicationConfig.WEBSOCKET_REQUIRE_SAME_ORIGIN, "false");
        String origins = System.getenv().getOrDefault("ALLOWED_ORIGINS", "https://chat.example.test");
        atmosphere.setInitParameter("chatterbox.allowedOrigins", origins);
        atmosphere.setInitParameter(ApplicationConfig.CORS_ALLOWED_ORIGINS, origins);
        atmosphere.setInitParameter(ApplicationConfig.ATMOSPHERE_INTERCEPTORS, OriginInterceptor.class.getName());
        // Treat a close frame without status code (1005) as a disconnect so @Disconnect fires.
        atmosphere.setInitParameter(ApplicationConfig.ALLOW_WEBSOCKET_STATUS_CODE_1005_AS_DISCONNECT, "true");
        // Atmosphere's own heartbeat writes a padding frame that is not Chatterbox JSON: switch it off.
        atmosphere.setInitParameter(ApplicationConfig.HEARTBEAT_INTERVAL_IN_SECONDS, "0");
        atmosphere.setInitOrder(1);
        atmosphere.setAsyncSupported(true);
        context.addServlet(atmosphere, "/*");
        ServletHolder tickets = new ServletHolder(new Tickets());
        tickets.setInitParameter("allowedOrigins", origins);
        context.addServlet(tickets, "/tickets");

        server.setHandler(context);
        server.start();
        server.join();
    }
}

package example.chatterbox;

import jakarta.servlet.http.HttpServlet;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * One-time tickets that the first hello frame redeems. A real application issues
 * a ticket from an authenticated HTTP endpoint; this demo trusts the query string.
 */
public final class Tickets extends HttpServlet {
    record Ticket(String userId, String name, long expiresAt) {
    }

    private static final long TTL_MS = 30_000;
    private static final Map<String, Ticket> TICKETS = new ConcurrentHashMap<>();

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
        String user = req.getParameter("user");
        if (user == null || !user.matches("[a-z0-9]{1,32}")) {
            resp.sendError(400, "user must match [a-z0-9]{1,32}");
            return;
        }
        String origin = req.getHeader("Origin");
        if (origin != null && getInitParameter("allowedOrigins") != null
                && List.of(getInitParameter("allowedOrigins").split(",")).contains(origin)) {
            resp.setHeader("Access-Control-Allow-Origin", origin); // the page may live on another origin
            resp.setHeader("Vary", "Origin");
        }
        String ticket = UUID.randomUUID().toString();
        TICKETS.put(ticket, new Ticket(user, req.getParameter("name") == null ? user : req.getParameter("name"),
                System.currentTimeMillis() + TTL_MS));
        resp.setContentType("application/json");
        resp.getWriter().write("{\"ticket\":\"" + ticket + "\"}");
    }

    /** Redeems a ticket once; returns null when it is unknown, used or expired. */
    static Ticket redeem(String ticket) {
        Ticket t = ticket == null ? null : TICKETS.remove(ticket);
        return t != null && t.expiresAt() > System.currentTimeMillis() ? t : null;
    }
}

package example.chatterbox;

import java.util.Arrays;
import java.util.Set;
import java.util.stream.Collectors;
import org.atmosphere.cpr.Action;
import org.atmosphere.cpr.AtmosphereConfig;
import org.atmosphere.cpr.AtmosphereInterceptorAdapter;
import org.atmosphere.cpr.AtmosphereResource;
import org.atmosphere.websocket.WebSocket;

/** Allow-list for the Origin header; requests without Origin (non-browsers) pass and rely on the ticket. */
public final class OriginInterceptor extends AtmosphereInterceptorAdapter {
    private Set<String> allowed = Set.of();

    @Override
    public void configure(AtmosphereConfig config) {
        String list = config.getInitParameter("chatterbox.allowedOrigins", "");
        allowed = Arrays.stream(list.split(",")).map(String::trim).filter(s -> !s.isEmpty())
                .collect(Collectors.toSet());
    }

    @Override
    public PRIORITY priority() {
        return PRIORITY.FIRST_BEFORE_DEFAULT;
    }

    @Override
    public Action inspect(AtmosphereResource r) {
        String origin = r.getRequest().getHeader("Origin");
        if (origin == null || allowed.contains(origin)) {
            return Action.CONTINUE;
        }
        if (r.getResponse().getAsyncIOWriter() instanceof WebSocket ws) {
            ws.close(4403, "origin not allowed");
        }
        return Action.CANCELLED;
    }
}

package example.chatterbox;

import java.security.SecureRandom;

/** Server-originated envelope ids: "s_" followed by a 26-character ULID. */
final class Ids {
    private static final char[] ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ".toCharArray();
    private static final SecureRandom RANDOM = new SecureRandom();

    private Ids() {
    }

    static String serverId() {
        return "s_" + ulid();
    }

    static String ulid() {
        char[] out = new char[26];
        long time = System.currentTimeMillis();
        for (int i = 9; i >= 0; i--) {
            out[i] = ALPHABET[(int) (time & 31)];
            time >>>= 5;
        }
        for (int i = 10; i < 26; i++) {
            out[i] = ALPHABET[RANDOM.nextInt(32)];
        }
        return new String(out);
    }
}

package example.chatterbox;

import jakarta.inject.Inject;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.atmosphere.config.service.Disconnect;
import org.atmosphere.config.service.ManagedService;
import org.atmosphere.config.service.Message;
import org.atmosphere.config.service.Ready;
import org.atmosphere.config.service.Singleton;
import org.atmosphere.cpr.AtmosphereResource;
import org.atmosphere.cpr.AtmosphereResourceEvent;
import org.atmosphere.cpr.Broadcaster;
import org.atmosphere.cpr.BroadcasterFactory;
import org.atmosphere.cpr.RawMessage;
import org.atmosphere.websocket.WebSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tools.jackson.core.JacksonException;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.json.JsonMapper;
import tools.jackson.databind.node.ArrayNode;
import tools.jackson.databind.node.ObjectNode;

/** Chatterbox wire protocol v1 on one Atmosphere endpoint: one socket, many rooms. */
@Singleton
@ManagedService(path = "/ws")
public class ChatterboxEndpoint {
    private static final Logger log = LoggerFactory.getLogger(ChatterboxEndpoint.class);
    private static final ObjectMapper JSON = JsonMapper.builder().build();
    private static final int MAX_FRAME_BYTES = 16384;
    private static final int HEARTBEAT_MS = Integer.getInteger("chatterbox.heartbeatMs", 20_000);
    private static final long GRACE_MS = Long.getLong("chatterbox.graceMs", 15_000);
    private static final int ACK_CACHE = 1000;

    /** Per-connection state, keyed by the AtmosphereResource uuid. */
    private static final class Conn {
        final AtmosphereResource resource;
        String userId;
        String name;
        String status = "online";
        volatile boolean helloDone;
        volatile long lastSeen = System.currentTimeMillis();
        double tokens = 40;
        long refilled = System.nanoTime();
        int rateErrors;
        long firstRateError;
        final Set<String> rooms = ConcurrentHashMap.newKeySet();

        Conn(AtmosphereResource resource) {
            this.resource = resource;
        }
    }

    private static final class Room {
        final String name;
        final Broadcaster broadcaster;
        long seq;
        final List<ObjectNode> events = new ArrayList<>();
        final Map<String, ObjectNode> messages = new LinkedHashMap<>();
        final Set<String> reactions = new HashSet<>();
        final Set<Conn> conns = ConcurrentHashMap.newKeySet();

        Room(String name, Broadcaster broadcaster) {
            this.name = name;
            this.broadcaster = broadcaster;
        }
    }

    private final Map<String, Conn> conns = new ConcurrentHashMap<>();
    private final Map<String, Room> rooms = new ConcurrentHashMap<>();
    private final Map<String, Map<String, ObjectNode>> acked = new ConcurrentHashMap<>();
    private final ScheduledExecutorService timers = Executors.newSingleThreadScheduledExecutor();

    @Inject
    private BroadcasterFactory factory;

    public ChatterboxEndpoint() {
        timers.scheduleAtFixedRate(this::sweep, HEARTBEAT_MS, HEARTBEAT_MS, TimeUnit.MILLISECONDS);
    }

    @Ready
    public void onReady(AtmosphereResource r) {
        Conn c = new Conn(r);
        conns.put(r.uuid(), c);
        timers.schedule(() -> {
            if (!c.helloDone && conns.containsKey(r.uuid())) {
                close(c, 4408, "hello timeout");
            }
        }, 5, TimeUnit.SECONDS);
    }

    @Disconnect
    public void onDisconnect(AtmosphereResourceEvent event) {
        log.debug("disconnect {} cancelled={} closedByClient={}", event.getResource().uuid(), event.isCancelled(), event.isClosedByClient());
        Conn c = conns.remove(event.getResource().uuid());
        if (c == null) {
            return;
        }
        c.rooms.forEach(name -> rooms.get(name).conns.remove(c));
        if (c.userId != null) {
            timers.schedule(() -> goOfflineIfGone(c), GRACE_MS, TimeUnit.MILLISECONDS);
        }
    }

    @Message
    public void onFrame(AtmosphereResource r, String frame) {
        Conn c = conns.get(r.uuid());
        if (c == null) {
            return;
        }
        c.lastSeen = System.currentTimeMillis();
        if (frame.getBytes(StandardCharsets.UTF_8).length > MAX_FRAME_BYTES) {
            close(c, 4413, "frame too large");
            return;
        }
        JsonNode in;
        try {
            in = JSON.readTree(frame);
        } catch (JacksonException e) {
            error(c, null, null, "bad_request", "frame is not JSON", false);
            return;
        }
        String id = text(in, "id");
        String type = text(in, "type");
        String roomName = in.path("room").isString() ? in.path("room").stringValue("") : null;
        JsonNode payload = in.path("payload");
        if (!in.path("v").isInt() || !payload.isObject() || type == null || id == null) {
            error(c, id, roomName, "bad_request", "v, type, id and payload are required", false);
            return;
        }
        if (in.path("v").intValue() != 1) {
            error(c, id, roomName, "unsupported_version", "only v1 is supported", false);
            close(c, 4426, "unsupported version");
            return;
        }
        if (!takeToken(c)) {
            error(c, id, roomName, "rate_limited", "slow down", true);
            return;
        }
        if (!c.helloDone && !type.equals("hello")) {
            error(c, id, roomName, "unauthorized", "send hello first", false);
            close(c, 4401, "unauthorized");
            return;
        }
        Map<String, ObjectNode> cache = c.userId == null ? null
                : acked.computeIfAbsent(c.userId, k -> new LinkedHashMap<>());
        if (cache != null) {
            ObjectNode previous;
            synchronized (cache) {
                previous = cache.get(id);
            }
            if (previous != null) {
                ObjectNode copy = previous.deepCopy();
                ((ObjectNode) copy.get("payload")).put("dup", true);
                write(c, copy);
                return;
            }
        }
        switch (type) {
            case "hello" -> hello(c, id, payload);
            case "join" -> join(c, id, roomName, payload);
            case "leave" -> leave(c, id, roomName);
            case "message" -> message(c, id, roomName, payload);
            case "edit" -> edit(c, id, roomName, payload);
            case "react" -> react(c, id, roomName, payload);
            case "typing" -> relay(c, "typing", roomName, JSON.createObjectNode()
                    .put("userId", c.userId).put("state", payload.path("state").asString("stop")));
            case "receipt" -> receipt(c, id, roomName, payload);
            case "presence" -> presence(c, id, payload);
            case "history" -> history(c, id, roomName, payload);
            default -> error(c, id, roomName, "bad_request", "unknown type " + type, false);
        }
    }

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

    private void hello(Conn c, String id, JsonNode p) {
        if (c.helloDone) {
            error(c, id, null, "conflict", "already said hello", false);
            return;
        }
        Tickets.Ticket t = Tickets.redeem(p.path("token").asString(null));
        if (t == null) {
            error(c, id, null, "unauthorized", "invalid or expired ticket", false);
            close(c, 4401, "unauthorized");
            return;
        }
        c.userId = t.userId();
        c.name = t.name();
        c.helloDone = true;
        ObjectNode reply = JSON.createObjectNode().put("ref", id).put("userId", c.userId).put("name", c.name)
                .put("sessionId", Ids.ulid()).put("heartbeatMs", HEARTBEAT_MS).put("maxFrameBytes", MAX_FRAME_BYTES)
                .put("resumed", false);
        reply.putArray("rooms");
        write(c, envelope("hello", null, reply));
    }

    private void join(Conn c, String id, String roomName, JsonNode p) {
        if (!validRoom(c, id, roomName)) {
            return;
        }
        Room room = rooms.computeIfAbsent(roomName, n -> new Room(n, factory.lookup("/rooms/" + n, true)));
        boolean first = c.rooms.add(roomName);
        room.conns.add(c);
        room.broadcaster.addAtmosphereResource(c.resource);
        ObjectNode result = JSON.createObjectNode();
        ArrayNode members = result.putArray("members");
        Set<String> seen = new HashSet<>();
        for (Conn m : room.conns) {
            if (seen.add(m.userId)) {
                members.addObject().put("userId", m.userId).put("name", m.name).put("status", m.status);
            }
        }
        long lastSeq;
        synchronized (room) {
            lastSeq = room.seq;
            result.put("lastSeq", lastSeq);
            ack(c, id, roomName, null, false, result);
            if (p.path("since").isIntegralNumber()) {
                long since = p.path("since").longValue();
                room.events.stream().filter(e -> e.path("payload").path("seq").longValue() > since)
                        .forEach(e -> write(c, e));
            }
        }
        if (first) {
            relay(c, "presence", roomName, JSON.createObjectNode().put("userId", c.userId).put("name", c.name)
                    .put("status", c.status));
        }
    }

    private void leave(Conn c, String id, String roomName) {
        Room room = roomName == null ? null : rooms.get(roomName);
        if (room == null || !c.rooms.remove(roomName)) {
            error(c, id, roomName, "not_found", "not a member of " + roomName, false);
            return;
        }
        room.conns.remove(c);
        room.broadcaster.removeAtmosphereResource(c.resource);
        ack(c, id, roomName, null, false, JSON.createObjectNode());
        relay(c, "presence", roomName, JSON.createObjectNode().put("userId", c.userId).put("name", c.name)
                .put("status", "offline"));
    }

    private void message(Conn c, String id, String roomName, JsonNode p) {
        Room room = memberRoom(c, id, roomName);
        if (room == null) {
            return;
        }
        String text = p.path("text").asString(null);
        if (text == null || text.isEmpty() || text.length() > 4000) {
            error(c, id, roomName, "bad_request", "text must be 1..4000 characters", false);
            return;
        }
        if (p.path("attachments").isArray() && p.path("attachments").size() > 5) {
            error(c, id, roomName, "bad_request", "at most 5 attachments", false);
            return;
        }
        synchronized (room) {
            String replyTo = p.path("replyTo").isString() ? p.path("replyTo").stringValue("") : null;
            if (replyTo != null && !room.messages.containsKey(replyTo)) {
                error(c, id, roomName, "not_found", "replyTo is not a message of this room", false);
                return;
            }
            long seq = ++room.seq;
            ObjectNode body = JSON.createObjectNode();
            body.putObject("from").put("userId", c.userId).put("name", c.name);
            body.put("text", text);
            body.set("replyTo", replyTo == null ? JSON.nullNode() : JSON.valueToTree(replyTo));
            body.set("attachments", p.path("attachments").isArray() ? p.path("attachments") : JSON.createArrayNode());
            body.put("seq", seq).put("edited", false);
            ObjectNode env = envelope("message", roomName, body);
            env.put("id", id);
            room.events.add(env);
            room.messages.put(id, env);
            room.broadcaster.broadcast(new RawMessage(JSON.writeValueAsString(env)));
            ack(c, id, roomName, seq, false, JSON.createObjectNode());
        }
    }

    private void edit(Conn c, String id, String roomName, JsonNode p) {
        Room room = memberRoom(c, id, roomName);
        if (room == null) {
            return;
        }
        synchronized (room) {
            ObjectNode target = room.messages.get(p.path("target").asString(""));
            if (target == null) {
                error(c, id, roomName, "not_found", "unknown target", false);
                return;
            }
            if (!target.path("payload").path("from").path("userId").asString("").equals(c.userId)) {
                error(c, id, roomName, "forbidden", "only the author may edit", false);
                return;
            }
            String text = p.path("text").asString("");
            if (text.isEmpty() || text.length() > 4000) {
                error(c, id, roomName, "bad_request", "text must be 1..4000 characters", false);
                return;
            }
            long seq = ++room.seq;
            ((ObjectNode) target.get("payload")).put("text", text).put("edited", true);
            ObjectNode body = JSON.createObjectNode().put("target", p.path("target").asString()).put("text", text)
                    .put("seq", seq).put("editedAt", System.currentTimeMillis());
            ObjectNode env = envelope("edit", roomName, body);
            env.put("id", id);
            room.events.add(env);
            room.broadcaster.broadcast(new RawMessage(JSON.writeValueAsString(env)));
            ack(c, id, roomName, seq, false, JSON.createObjectNode());
        }
    }

    private void react(Conn c, String id, String roomName, JsonNode p) {
        Room room = memberRoom(c, id, roomName);
        if (room == null) {
            return;
        }
        synchronized (room) {
            String target = p.path("target").asString("");
            if (!room.messages.containsKey(target)) {
                error(c, id, roomName, "not_found", "unknown target", false);
                return;
            }
            String emoji = p.path("emoji").asString("");
            String op = p.path("op").asString("add");
            String key = target + "|" + c.userId + "|" + emoji;
            if (emoji.isEmpty() || !(op.equals("add") || op.equals("remove"))) {
                error(c, id, roomName, "bad_request", "emoji and op (add or remove) are required", false);
                return;
            }
            if (op.equals("add") ? !room.reactions.add(key) : !room.reactions.remove(key)) {
                error(c, id, roomName, op.equals("add") ? "conflict" : "not_found",
                        op.equals("add") ? "already reacted" : "no such reaction", false);
                return;
            }
            long seq = ++room.seq;
            ObjectNode body = JSON.createObjectNode().put("target", target).put("emoji", emoji).put("op", op);
            body.putObject("from").put("userId", c.userId);
            body.put("seq", seq);
            ObjectNode env = envelope("react", roomName, body);
            env.put("id", id);
            room.events.add(env);
            room.broadcaster.broadcast(new RawMessage(JSON.writeValueAsString(env)));
            ack(c, id, roomName, seq, false, JSON.createObjectNode());
        }
    }

    private void receipt(Conn c, String id, String roomName, JsonNode p) {
        Room room = roomName == null ? null : rooms.get(roomName);
        if (room == null || !c.rooms.contains(roomName)) {
            return;
        }
        long upTo = p.path("upTo").longValue();
        if (upTo > room.seq) {
            error(c, id, roomName, "bad_request", "upTo is beyond the last seq", false);
            return;
        }
        relay(c, "receipt", roomName, JSON.createObjectNode().put("userId", c.userId)
                .put("kind", p.path("kind").asString("read")).put("upTo", upTo));
    }

    private void presence(Conn c, String id, JsonNode p) {
        String status = p.path("status").asString("online");
        if (!status.equals("online") && !status.equals("away")) {
            error(c, id, null, "bad_request", "status must be online or away", false);
            return;
        }
        boolean changed = !status.equals(c.status);
        c.status = status;
        ack(c, id, null, null, false, JSON.createObjectNode());
        if (changed) {
            for (String r : c.rooms) {
                relay(c, "presence", r, JSON.createObjectNode().put("userId", c.userId).put("name", c.name)
                        .put("status", status));
            }
        }
    }

    private void history(Conn c, String id, String roomName, JsonNode p) {
        Room room = memberRoom(c, id, roomName);
        if (room == null) {
            return;
        }
        int limit = Math.max(1, Math.min(100, p.path("limit").asInt(50)));
        long before = p.path("before").isIntegralNumber() ? p.path("before").longValue() : Long.MAX_VALUE;
        ObjectNode result = JSON.createObjectNode();
        ArrayNode page = result.putArray("messages");
        synchronized (room) {
            List<ObjectNode> older = room.messages.values().stream()
                    .filter(m -> m.path("payload").path("seq").longValue() < before).toList();
            int from = Math.max(0, older.size() - limit);
            older.subList(from, older.size()).forEach(page::add);
            boolean hasMore = from > 0;
            result.put("hasMore", hasMore);
            if (hasMore) {
                result.put("next", older.get(from).path("payload").path("seq").longValue());
            } else {
                result.putNull("next");
            }
        }
        ack(c, id, roomName, null, false, result);
    }

    // ---- plumbing ----------------------------------------------------------------------------

    private boolean validRoom(Conn c, String id, String roomName) {
        if (roomName == null || !roomName.matches("[a-z0-9-]{1,64}")) {
            error(c, id, roomName, "bad_request", "room must match [a-z0-9-]{1,64}", false);
            return false;
        }
        if (roomName.startsWith("dm-") && !List.of(roomName.substring(3).split("-")).contains(c.userId)) {
            error(c, id, roomName, "forbidden", "not a party of this DM", false);
            return false;
        }
        return true;
    }

    private Room memberRoom(Conn c, String id, String roomName) {
        Room room = roomName == null ? null : rooms.get(roomName);
        if (room == null || !c.rooms.contains(roomName)) {
            error(c, id, roomName, "forbidden", "join the room first", false);
            return null;
        }
        return room;
    }

    /** Sends typing, receipt and presence envelopes to the other members of a room. */
    private void relay(Conn from, String type, String roomName, ObjectNode body) {
        Room room = rooms.get(roomName);
        if (room == null) {
            return;
        }
        Set<AtmosphereResource> others = new HashSet<>();
        room.conns.stream().filter(m -> m != from).forEach(m -> others.add(m.resource));
        if (!others.isEmpty()) {
            room.broadcaster.broadcast(new RawMessage(JSON.writeValueAsString(envelope(type, roomName, body))), others);
        }
    }

    private void goOfflineIfGone(Conn gone) {
        boolean stillHere = conns.values().stream().anyMatch(m -> gone.userId.equals(m.userId));
        if (!stillHere) {
            gone.rooms.forEach(r -> {
                Room room = rooms.get(r);
                if (room != null) {
                    ObjectNode body = JSON.createObjectNode().put("userId", gone.userId).put("name", gone.name)
                            .put("status", "offline");
                    room.broadcaster.broadcast(new RawMessage(JSON.writeValueAsString(envelope("presence", r, body))));
                }
            });
        }
    }

    private void sweep() {
        long now = System.currentTimeMillis();
        for (Conn c : conns.values()) {
            if (now - c.lastSeen > 2L * HEARTBEAT_MS) {
                close(c, 4408, "heartbeat timeout");
            } else if (c.resource.getResponse().getAsyncIOWriter() instanceof WebSocket ws) {
                try {
                    ws.sendPing(new byte[0]);
                } catch (RuntimeException e) {
                    log.debug("ping failed", e);
                }
            }
        }
    }

    private boolean takeToken(Conn c) {
        long now = System.nanoTime();
        c.tokens = Math.min(40, c.tokens + (now - c.refilled) / 1e9 * 20);
        c.refilled = now;
        if (c.tokens >= 1) {
            c.tokens -= 1;
            return true;
        }
        long ms = System.currentTimeMillis();
        if (ms - c.firstRateError > 10_000) {
            c.firstRateError = ms;
            c.rateErrors = 0;
        }
        if (++c.rateErrors >= 10) {
            close(c, 4429, "rate limited");
        }
        return false;
    }

    private static String text(JsonNode n, String field) {
        return n.path(field).isString() ? n.path(field).stringValue("") : null;
    }

    private static ObjectNode envelope(String type, String room, ObjectNode payload) {
        ObjectNode env = JSON.createObjectNode().put("v", 1).put("type", type).put("id", Ids.serverId());
        if (room == null) {
            env.putNull("room");
        } else {
            env.put("room", room);
        }
        env.put("ts", System.currentTimeMillis());
        env.set("payload", payload);
        return env;
    }

    private void ack(Conn c, String ref, String room, Long seq, boolean dup, ObjectNode result) {
        ObjectNode body = JSON.createObjectNode().put("ref", ref).put("ok", true);
        if (seq == null) {
            body.putNull("seq");
        } else {
            body.put("seq", seq.longValue());
        }
        body.put("dup", dup);
        body.set("result", result);
        ObjectNode env = envelope("ack", room, body);
        Map<String, ObjectNode> cache = acked.computeIfAbsent(c.userId, k -> new LinkedHashMap<>());
        synchronized (cache) {
            cache.put(ref, env.deepCopy());
            if (cache.size() > ACK_CACHE) {
                cache.remove(cache.keySet().iterator().next());
            }
        }
        write(c, env);
    }

    private void error(Conn c, String ref, String room, String code, String message, boolean retryable) {
        ObjectNode body = JSON.createObjectNode();
        if (ref == null) {
            body.putNull("ref");
        } else {
            body.put("ref", ref);
        }
        body.put("code", code).put("message", message).put("retryable", retryable);
        write(c, envelope("error", room, body));
    }

    private void write(Conn c, JsonNode env) {
        try {
            c.resource.write(JSON.writeValueAsString(env));
        } catch (RuntimeException e) {
            log.debug("write to {} failed: {}", c.resource.uuid(), e.getMessage());
        }
    }

    /** Closes with an application code after a short delay, so the error frame written just before drains first. */
    private void close(Conn c, int code, String reason) {
        timers.schedule(() -> {
            if (c.resource.getResponse().getAsyncIOWriter() instanceof WebSocket ws) {
                ws.close(code, reason);
            }
        }, 100, TimeUnit.MILLISECONDS);
    }
}

docker run --rm -v "$PWD":/app -v "$HOME/.m2":/root/.m2 -w /app maven:3.9.16-eclipse-temurin-21 \
  mvn -q -B package -DskipTests

docker run --rm -p 8080:8080 -v "$PWD":/app -w /app \
  -e ALLOWED_ORIGINS=https://chat.example.test \
  eclipse-temurin:21.0.12.1_1-jre java -cp 'target/classes:target/lib/*' example.chatterbox.Main

// demo.mjs - drives the Chatterbox server: join, broadcast, ack, duplicate id, error.
// Needs Node.js 22 or later (global fetch, crypto) and the ws package: npm install ws@8.22.0
import WebSocket from 'ws';

const BASE = process.env.CHATTERBOX_URL ?? 'http://localhost:8080';
const ORIGIN = 'https://chat.example.test';
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const newId = () => crypto.randomUUID().replaceAll('-', '').slice(0, 26).toUpperCase();

async function connect(user) {
  const { ticket } = await (await fetch(`${BASE}/tickets?user=${user}`, { method: 'POST' })).json();
  const ws = new WebSocket(BASE.replace('http', 'ws') + '/ws', { origin: ORIGIN });
  const inbox = [];
  ws.on('message', (data) => inbox.push(JSON.parse(data.toString())));
  await new Promise((resolve) => ws.on('open', resolve));

  const client = {
    ws,
    send(type, room, payload, id = newId()) {
      ws.send(JSON.stringify({ v: 1, type, id, room, ts: 0, payload }));
      return id;
    },
    async next(predicate) {
      for (let i = 0; i < 150; i++) {
        const at = inbox.findIndex(predicate);
        if (at >= 0) return inbox.splice(at, 1)[0];
        await sleep(20);
      }
      throw new Error(`${user}: timed out waiting, inbox=${JSON.stringify(inbox)}`);
    },
    // One request, one answer: the ack or the error whose ref is the id we sent.
    async request(type, room, payload, id = newId()) {
      client.send(type, room, payload, id);
      return client.next((e) => (e.type === 'ack' || e.type === 'error') && e.payload.ref === id);
    },
  };
  client.send('hello', null, { token: ticket, client: 'demo/1.0' });
  await client.next((e) => e.type === 'hello');
  return client;
}

const line = (who, text) => console.log(`${who.padEnd(6)} ${text}`);
const alice = await connect('alice');
const bob = await connect('bob');
const room = 'general-' + Date.now().toString(36);

let r = await alice.request('join', room, { since: null });
line('alice', `join -> ${r.type} lastSeq=${r.payload.result.lastSeq}`);
r = await bob.request('join', room, { since: null });
line('bob', `join -> ${r.type} members=${r.payload.result.members.map((m) => m.userId).sort()}`);
const presence = await alice.next((e) => e.type === 'presence');
line('alice', `presence <- ${presence.payload.userId} is ${presence.payload.status}`);

const id = newId();
r = await alice.request('message', room, { text: 'Hello, Chatterbox', replyTo: null, attachments: [] }, id);
line('alice', `message -> ack seq=${r.payload.seq} dup=${r.payload.dup}`);
for (const who of [alice, bob]) {
  const m = await who.next((e) => e.type === 'message');
  line(who === alice ? 'alice' : 'bob', `message <- "${m.payload.text}" from ${m.payload.from.userId} seq=${m.payload.seq}`);
}

r = await alice.request('message', room, { text: 'Hello, Chatterbox', replyTo: null, attachments: [] }, id);
line('alice', `same id again -> ack seq=${r.payload.seq} dup=${r.payload.dup} (no second broadcast)`);

r = await bob.request('edit', room, { target: id, text: 'not my message' });
line('bob', `edit of alice's message -> ${r.type} ${r.payload.code} (${r.payload.message})`);

r = await bob.request('message', 'other-room', { text: 'hi' });
line('bob', `message to a room never joined -> ${r.type} ${r.payload.code}`);

r = await bob.request('history', room, { before: null, limit: 10 });
line('bob', `history -> ${r.payload.result.messages.length} message(s), hasMore=${r.payload.result.hasMore}`);

alice.ws.close(1000);
bob.ws.close(1000);
await sleep(200);

alice  join -> ack lastSeq=0
bob    join -> ack members=alice,bob
alice  presence <- bob is online
alice  message -> ack seq=1 dup=false
alice  message <- "Hello, Chatterbox" from alice seq=1
bob    message <- "Hello, Chatterbox" from alice seq=1
alice  same id again -> ack seq=1 dup=true (no second broadcast)
bob    edit of alice's message -> error forbidden (only the author may edit)
bob    message to a room never joined -> error forbidden
bob    history -> 1 message(s), hasMore=false

<!doctype html>
<meta charset="utf-8">
<title>Chatterbox on Atmosphere</title>
<script>
  const q = new URLSearchParams(location.search);
  const API = q.get('api') || 'http://localhost:8210';
  window.__log = [];
  window.__closed = null;

  async function connect(user) {
    const { ticket } = await (await fetch(`${API}/tickets?user=${user}&name=${user}`, { method: 'POST' })).json();
    // No subprotocol argument: Atmosphere's JSR 356 endpoint cannot select one, so the browser would refuse the socket.
    const ws = new WebSocket(API.replace('http', 'ws') + '/ws');
    const pending = new Map();
    ws.onmessage = (ev) => {
      const env = JSON.parse(ev.data);
      __log.push(env);
      const ref = env.payload && env.payload.ref;
      if (pending.has(ref) && (env.type === 'ack' || env.type === 'error' || env.type === 'hello')) {
        pending.get(ref)(env);
        pending.delete(ref);
      }
    };
    ws.onclose = (ev) => { __closed = ev.code; };
    const send = (type, room, payload) => new Promise((resolve) => {
      const id = crypto.randomUUID().replaceAll('-', '').slice(0, 26).toUpperCase();
      pending.set(id, resolve);
      ws.send(JSON.stringify({ v: 1, type, id, room, ts: 0, payload }));
    });
    await new Promise((resolve, reject) => { ws.onopen = resolve; ws.onerror = () => reject(new Error('socket failed')); });
    const hello = await send('hello', null, { token: ticket, client: 'web/1.4' });
    return { ws, send, hello };
  }
  window.connect = connect;
</script>