Feat: pulse users clustering

proto/decentraland/pulse/pulse_clusters.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+package decentraland.pulse;
+
+// Published by Pulse to NATS on `peer.{address}.cluster_change` whenever a peer's post-debounce
+// cluster assignment changes. Pulse is the sole author of cluster membership.
+//
+// A "cluster" is Pulse's own grouping of co-located peers, derived by union-find over occupied
+// area-of-interest grid cells. It is deliberately a different concept from archipelago's "island":
+// clusters have no size cap, are realm-partitioned, and carry no transport connection details.
+// Comms Gatekeeper subscribes here, mints the LiveKit connection string (the ban check is
+// authority-local to gatekeeper), and re-emits the legacy
+// decentraland.kernel.comms.v3.IslandChangedMessage on `engine.peer.{address}.island_changed`,
+// so the WebSocket Connector and clients stay untouched.
+message PeerClusterChange {
+  // Cluster the peer now belongs to. Stable across passes for an unchanged crowd.
+  string cluster_id = 1;
+
+  // The peer's realm. Clusters never span realms, so this qualifies cluster_id and keeps the
+  // event self-contained — consumers never have to join against the topology feed.
+  string realm = 2;
+}

proto/decentraland/pulse/pulse_presence.proto

@@ -0,0 +1,41 @@
+syntax = "proto3";
+
+package decentraland.pulse;
+
+// Published by Pulse to NATS on `engine.parcel_changes`: the platform's single presence feed.
+// One entry per (peer, realm) state change: the first non-null placement in a realm is the
+// connect signal, an entry with `parcel` absent is the exit signal for that realm/instance,
+// and every entry carries the realm and the parcel the peer now stands on.
+//
+// Guarantees the producer (Pulse) holds — see archipelago-workers/docs/contracts/iteration-2/README.md:
+//   1. A peer's first placement in a realm emits one non-null entry, even if it never moves afterwards.
+//   2. Every exit path emits exactly one parcel-absent entry (clean disconnect, PENDING_AUTH / idle
+//      timeout, duplicate-session kick, BanEnforcer eviction, PeerDefense kick). A realm change
+//      emits a non-null entry for the new realm only; the old realm is implied.
+//   3. Within one batch a wallet appears at most once, with its latest state.
+//   4. Cadence: every Presence:BatchIntervalMs (default 2000) when non-empty; a snapshot=true batch
+//      on publisher start, every Presence:SnapshotIntervalMs (default 60000), and immediately after
+//      any outbox eviction, so consumers never run on a known-lossy delta stream.
+//   5. `realm` and `address` are lowercase; `server_name` is stable for the process lifetime.
+//
+// Consumer rule: keep lastSeq[server_name]; on a gap keep serving stale state and wait for the next
+// snapshot (<= 60 s) — never drop the map. On snapshot=true replace every entry whose server_name matches.
+
+message Parcel {
+  int32 x = 1;
+  int32 y = 2;
+}
+
+message ParcelChange {
+  string address = 1;          // lowercase 0x wallet
+  string realm   = 2;          // canonical lowercase
+  optional Parcel parcel = 3;  // absent => peer left this realm/instance
+}
+
+message ParcelChangesBatch {
+  string server_name = 1;      // NatsOptions.ServerName; consumers key seq per server_name
+  uint64 seq         = 2;      // monotonic per server_name, resets on restart (snapshot=true follows)
+  bool   snapshot    = 3;      // true => full state of this server, replaces everything known for server_name
+  uint64 server_time = 4;      // unix ms
+  repeated ParcelChange changes = 5;
+}

public/comms.proto

@@ -2,4 +2,6 @@ syntax = "proto3";
 
 import public "decentraland/kernel/comms/v3/archipelago.proto";
 import public "decentraland/kernel/comms/rfc4/comms.proto";
-import public "decentraland/kernel/comms/rfc5/ws_comms.proto";
+import public "decentraland/kernel/comms/rfc5/ws_comms.proto";
+import public "decentraland/pulse/pulse_clusters.proto";
+import public "decentraland/pulse/pulse_presence.proto";