Emit presence on identify, support deduped gw events

src/gateway/listener/listener.ts

@@ -205,6 +205,13 @@ async function consume(this: WebSocket, opts: EventOpts) {
     opts.acknowledge?.();
     // console.log("event", event);
 
+    // deduplicate gateway messages
+    if (opts.transaction_id) {
+        if (this.recentTransactions.includes(opts.transaction_id)) return;
+        this.recentTransactions.push(opts.transaction_id);
+        if (this.recentTransactions.length > 100) this.recentTransactions = this.recentTransactions.slice(1);
+    }
+
     // special codes
     switch (event) {
         case "SB_SESSION_CLOSE":

src/gateway/opcodes/Identify.ts

@@ -61,6 +61,7 @@ import { check } from "./instanceOf";
 import { In, Not } from "typeorm";
 import { PreloadedUserSettings } from "discord-protos";
 import { ChannelType, DefaultUserGuildSettings, DMChannel, IdentifySchema, PrivateUserProjection, PublicUser, PublicUserProjection } from "@spacebar/schemas";
+import { randomString } from "@spacebar/api*";
 
 // TODO: user sharding
 // TODO: check privileged intents, if defined in the config
@@ -162,7 +163,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
 
     this.session_id = session.session_id;
     this.session = session;
-    this.session.status = identify.presence?.status || "online";
+    // this.session.status = identify.presence?.status || "online";
     this.session.last_seen = new Date();
     this.session.client_info ??= {};
     this.session.client_info.platform = identify.properties?.$device ?? identify.properties?.$device;
@@ -175,6 +176,33 @@ export async function onIdentify(this: WebSocket, data: Payload) {
         await this.session.updateIpInfo();
     }
 
+    let mustAnnouncePresence = false;
+    let presenceUpdateEventData: PresenceUpdateEvent | undefined;
+
+    if (identify.presence?.status) {
+        let newStatus = identify.presence.status;
+        if (newStatus == "unknown") newStatus = this.session.status;
+        if (newStatus == "offline") {
+            newStatus = "online";
+            mustAnnouncePresence = true;
+        }
+
+        this.session.status = newStatus;
+        if (mustAnnouncePresence) {
+            presenceUpdateEventData = {
+                event: "PRESENCE_UPDATE",
+                data: {
+                    user: tokenData.user.toPublicUser(),
+                    status: this.session.status,
+                    client_status: this.session.client_status,
+                    activities: this.session.activities,
+                },
+                origin: "GATEWAY_IDENTIFY",
+                transaction_id: `IDENT_${this.user_id}_${randomString()}`,
+            } satisfies PresenceUpdateEvent;
+        }
+    }
+
     const createSessionTime = taskSw.getElapsedAndReset();
 
     // Get from database:
@@ -849,4 +877,30 @@ export async function onIdentify(this: WebSocket, data: Payload) {
         `[Gateway/${this.user_id}] IDENTIFY ${this.user_id} in ${totalSw.elapsed().totalMilliseconds}ms`,
         process.env.LOG_GATEWAY_TRACES ? JSON.stringify(d._trace, null, 2) : "",
     );
+
+    // actually send presence updates
+    if (presenceUpdateEventData) {
+        for (const rel of d.relationships ?? []) {
+            await emitEvent({
+                ...presenceUpdateEventData,
+                user_id: rel.user.id,
+            });
+        }
+        for (const guild of d.guilds) {
+            await emitEvent({
+                ...presenceUpdateEventData,
+                guild_id: guild.id,
+            });
+        }
+        for (const dmChannel of d.private_channels) {
+            // TODO: check if other side has the channel still open
+            for (const recpt of dmChannel.recipients) {
+                if (recpt.id != this.user_id)
+                    await emitEvent({
+                        ...presenceUpdateEventData,
+                        user_id: recpt.id,
+                    });
+            }
+        }
+    }
 }

src/gateway/util/SessionUtils.ts

@@ -37,6 +37,7 @@ export function getMostRelevantSession(sessions: Session[]) {
         dnd: 2,
         invisible: 3,
         offline: 4,
+        unknown: 5,
     };
     // sort sessions by relevance
     sessions = sessions.sort((a, b) => {

src/gateway/util/WebSocket.ts

@@ -24,6 +24,7 @@ import { Decoder, Encoder } from "@toondepauw/node-zstd";
 import { QoSPayload } from "../opcodes/Heartbeat";
 
 export interface WebSocket extends WS {
+    recentTransactions: string[];
     version: number;
     user_id: string;
     session_id: string;

src/util/interfaces/Event.ts

@@ -64,6 +64,7 @@ export interface Event {
     data?: any;
     reconnect_delay?: number;
     origin?: string;
+    transaction_id?: string;
 }
 
 // ! Custom Events that shouldn't get sent to the client but processed by the server

src/util/interfaces/Status.ts

@@ -16,11 +16,20 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
-export type Status = "idle" | "dnd" | "online" | "offline" | "invisible";
+export type Status =
+    | "idle"
+    | "dnd"
+    | "online"
+    | "offline"
+    // Send only
+    | "invisible"
+    // Identify only
+    | "unknown";
 
 export interface ClientStatus {
     desktop?: string; // e.g. Windows/Linux/Mac
     mobile?: string; // e.g. iOS/Android
     web?: string; // e.g. browser, bot account, unknown
     embedded?: string; // e.g. embedded
+    vr?: string;
 }