feat: server-side message sequencing, ring-buffer replay, and reconnection handshake

src/room-manager.js

@@ -49,6 +49,40 @@ export class RoomManager {
     }
   }
 
+  /** @private */
+  _isDuplicate(roomId, message, excludeClientId) {
+    let parsed = message;
+    if (typeof message === "string") {
+      try {
+        parsed = JSON.parse(message);
+      } catch {
+        return false;
+      }
+    }
+    if (parsed && typeof parsed === "object" && parsed.type === "location_update") {
+      const clientId = parsed.payload?.clientId || excludeClientId;
+      const timestamp = parsed.payload?.timestamp;
+      if (clientId && timestamp) {
+        const key = `${roomId}:${clientId}:${timestamp}`;
+        const now = Date.now();
+        if (this._dedupCache.has(key)) {
+          const recordedTime = this._dedupCache.get(key);
+          if (now - recordedTime <= this._deduplicationWindowMs) {
+            return true;
+          }
+        }
+        this._dedupCache.set(key, now);
+        if (this._dedupCache.size > this._maxDedupEntries) {
+          const oldestKey = this._dedupCache.keys().next().value;
+          if (oldestKey !== undefined) {
+            this._dedupCache.delete(oldestKey);
+          }
+        }
+      }
+    }
+    return false;
+  }
+
   join(clientId, roomId, ws) {
     if (clientId == null) throw new TypeError("clientId is required");
     if (roomId == null) throw new TypeError("roomId is required");
@@ -92,7 +126,7 @@ export class RoomManager {
 
   /**
    * Broadcasts a message to every open connection in a room, optionally
-   * excluding the sender.
+   * excluding the sender. Stores message in ring buffer with a sequence number.
    *
    * Objects are serialised to JSON; strings are sent as-is.
    * Clients whose `readyState` is not `OPEN` are silently skipped.
@@ -104,6 +138,45 @@ export class RoomManager {
   broadcast(roomId, message, excludeClientId = null) {
     if (roomId == null) throw new TypeError("roomId is required");
 
+    if (this._isDuplicate(roomId, message, excludeClientId)) {
+      return;
+    }
+
+    const currentSeq = (this._roomSeq.get(roomId) ?? 0) + 1;
+    this._roomSeq.set(roomId, currentSeq);
+
+    let payload = message;
+    if (typeof message === "string") {
+      try {
+        payload = JSON.parse(message);
+      } catch {
+        payload = message;
+      }
+    }
+
+    const entry = {
+      seq: currentSeq,
+      payload,
+      timestamp: Date.now(),
+    };
+
+    if (!this._roomBuffers.has(roomId)) {
+      this._roomBuffers.set(roomId, []);
+      this._roomBufferBytes.set(roomId, 0);
+    }
+
+    const buffer = this._roomBuffers.get(roomId);
+    const entryBytes = Buffer.byteLength(JSON.stringify(entry), "utf8");
+    let currentBytes = (this._roomBufferBytes.get(roomId) ?? 0) + entryBytes;
+    buffer.push(entry);
+
+    while (buffer.length > 0 && (buffer.length > this._ringBufferSize || currentBytes > this._maxBufferBytes)) {
+      const evicted = buffer.shift();
+      const evictedBytes = Buffer.byteLength(JSON.stringify(evicted), "utf8");
+      currentBytes -= evictedBytes;
+    }
+    this._roomBufferBytes.set(roomId, Math.max(0, currentBytes));
+
     const room = this._rooms.get(roomId);
     if (!room) return;
 
@@ -121,6 +194,72 @@ export class RoomManager {
     }
   }
 
+  /**
+   * Returns the current sequence number for a room.
+   *
+   * @param {string} roomId - Identifier of the room.
+   * @returns {number} Current sequence number, or 0 if room has no broadcasts.
+   */
+  getRoomSeq(roomId) {
+    if (roomId == null) throw new TypeError("roomId is required");
+    return this._roomSeq.get(roomId) ?? 0;
+  }
+
+  /**
+   * Returns the array of stored ring buffer entries for a room in order.
+   *
+   * @param {string} roomId - Identifier of the room.
+   * @returns {Array<{ seq: number, payload: any, timestamp: number }>}
+   */
+  getRingBuffer(roomId) {
+    if (roomId == null) throw new TypeError("roomId is required");
+    const buffer = this._roomBuffers.get(roomId);
+    return buffer ? [...buffer] : [];
+  }
+
+  /**
+   * Evaluates a reconnection request against stored room sequence and ring buffer.
+   *
+   * @param {string} roomId - Identifier of the target room.
+   * @param {number} lastSeq - The sequence number last received by the client.
+   * @returns {object} Replay payload (type: "replay", "replay_complete", or "replay_gap").
+   */
+  handleReconnect(roomId, lastSeq) {
+    if (roomId == null) throw new TypeError("roomId is required");
+    if (lastSeq == null) throw new TypeError("lastSeq is required");
+
+    const currentSeq = this.getRoomSeq(roomId);
+    const buffer = this._roomBuffers.get(roomId) ?? [];
+
+    if (lastSeq === currentSeq) {
+      return {
+        type: "replay_complete",
+        roomId,
+      };
+    }
+
+    if (buffer.length > 0) {
+      const oldestSeq = buffer[0].seq;
+      if (lastSeq >= oldestSeq - 1 && lastSeq < currentSeq) {
+        const missed = buffer.filter((entry) => entry.seq > lastSeq);
+        return {
+          type: "replay",
+          roomId,
+          messages: missed,
+          currentSeq,
+        };
+      }
+    }
+
+    const oldestSeq = buffer.length > 0 ? buffer[0].seq : currentSeq;
+    return {
+      type: "replay_gap",
+      roomId,
+      fromSeq: oldestSeq,
+      currentSeq,
+    };
+  }
+
   /**
    * Removes a client from all rooms they belong to and cleans up empty rooms.
    *

src/server.js

@@ -8,7 +8,17 @@ import { logger } from "./logger.js";
 import { createRateLimiter } from "./rate-limiter.js";
 import { createConnRateLimiter } from "./conn-rate-limiter.js";
 
-export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) {
+export function createServer({
+  port,
+  heartbeatMs,
+  maxPayloadBytes,
+  connRateLimit,
+  maxConnectionsPerIp,
+  ringBufferSize,
+  deduplicationWindowMs,
+  maxBufferBytes,
+  maxDedupEntries,
+} = {}) {
   const server = http.createServer((req, res) => {
     let url;
     try {
@@ -120,6 +130,16 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit
           ws.send(JSON.stringify({ type: "room_left", payload: { roomId: msg.roomId } }));
           break;
         }
+        case "reconnect": {
+          const clientRooms = rooms.getClientRooms(actualClientId);
+          if (!clientRooms.has(msg.roomId)) {
+            ws.send(JSON.stringify({ type: "error", payload: { message: "Must join room before reconnecting" } }));
+            break;
+          }
+          const replayResult = rooms.handleReconnect(msg.roomId, msg.lastSeq);
+          ws.send(JSON.stringify(replayResult));
+          break;
+        }
         case "location_update": {
           const roomIds = rooms.getClientRooms(actualClientId);
           for (const roomId of roomIds) {

src/validator.js

@@ -18,6 +18,11 @@ const leaveRoomSchema = z.object({
   roomId: z.string().min(1).max(128),
 });
 
+const reconnectSchema = z.object({
+  roomId: z.string().min(1).max(128),
+  lastSeq: z.number().min(0),
+});
+
 const messageSchema = z.discriminatedUnion("type", [
   z.object({
     type: z.literal("location_update"),
@@ -31,12 +36,17 @@ const messageSchema = z.discriminatedUnion("type", [
     type: z.literal("leave_room"),
     ...leaveRoomSchema.shape,
   }),
+  z.object({
+    type: z.literal("reconnect"),
+    ...reconnectSchema.shape,
+  }),
 ]);
 
 const MESSAGE_SIZE_LIMITS = {
   location_update: 512,
   join_room: 256,
   leave_room: 256,
+  reconnect: 256,
 };
 
 export function validateMessage(raw) {