gerbeur/api/services/ws-service.ts

import type {
  ChatMessage,
  Comment,
  Dump,
  OnlineUser,
  Playlist,
  Role,
  ServerToClientMessage,
  User,
} from "../model/interfaces.ts";

export interface WsClient {
  socket: WebSocket;
  userId?: string;
  username?: string;
  role?: Role;
  avatarMime?: string;
  avatarVersion?: number;
  pongReceived?: boolean;
  /** Whether this client currently has the chatbox open (is reading chat). */
  chatOpen?: boolean;
  /** Timestamp (ms) of the client's last "typing" signal; undefined when not
   * composing. Entries older than TYPING_TTL are treated as stale. */
  typingAt?: number;
}

const clients = new Set<WsClient>();

export function register(client: WsClient): void {
  clients.add(client);
}

export function unregister(client: WsClient): void {
  clients.delete(client);
}

export function updateClientAvatar(userId: string, avatarMime: string): void {
  const version = Date.now();
  for (const client of clients) {
    if (client.userId === userId) {
      client.avatarMime = avatarMime;
      client.avatarVersion = version;
    }
  }
  broadcastPresence();
}

export function getOnlineUsers(): OnlineUser[] {
  const seen = new Map<string, OnlineUser>();
  for (const client of clients) {
    if (client.userId && !seen.has(client.userId)) {
      seen.set(client.userId, {
        userId: client.userId,
        username: client.username!,
        hasAvatar: !!client.avatarMime,
        avatarVersion: client.avatarVersion,
      });
    }
  }
  return Array.from(seen.values());
}

function send(socket: WebSocket, data: ServerToClientMessage): void {
  if (socket.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify(data));
  }
}

export function sendToUser(userId: string, data: ServerToClientMessage): void {
  for (const client of clients) {
    if (client.userId === userId) {
      send(client.socket, data);
    }
  }
}

export function disconnectUser(userId: string): void {
  for (const client of clients) {
    if (client.userId === userId) {
      send(client.socket, { type: "force_logout" });
      client.socket.close(1000, "Account deleted");
    }
  }
}

/**
 * Drops the user's sockets without logging anyone out: each reconnects with
 * whatever token it holds, and /ws turns a revoked one into a force_logout.
 * Used on a password change, where the tab that made the change has already
 * swapped in a fresh token and must stay signed in.
 */
export function closeUserSockets(userId: string, reason: string): void {
  for (const client of clients) {
    if (client.userId === userId) client.socket.close(1000, reason);
  }
}

export function broadcastPresence(): void {
  const users = getOnlineUsers();
  for (const client of clients) {
    send(client.socket, { type: "presence_update", users });
  }
}

export function broadcastNewDump(dump: Dump): void {
  for (const client of clients) {
    send(client.socket, { type: "dump_created", dump });
  }
}

export function broadcastDumpUpdated(dump: Dump): void {
  for (const client of clients) {
    send(client.socket, { type: "dump_updated", dump });
  }
}

export function broadcastDumpDeleted(dumpId: string): void {
  for (const client of clients) {
    send(client.socket, { type: "dump_deleted", dumpId });
  }
}

// `onlyUserId`: a private dump's count goes to that user's sockets alone.
export function broadcastVoteUpdate(
  dumpId: string,
  voteCount: number,
  voterId: string,
  action: "cast" | "remove",
  onlyUserId?: string,
): void {
  for (const client of clients) {
    if (onlyUserId && client.userId !== onlyUserId) continue;
    send(client.socket, {
      type: "votes_update",
      dumpId,
      voteCount,
      voterId,
      action,
    });
  }
}

// `onlyUserId`: a like on a private dump's comment goes to that user alone.
export function broadcastCommentLikeUpdate(
  commentId: string,
  likeCount: number,
  likerId: string,
  action: "cast" | "remove",
  onlyUserId?: string,
): void {
  for (const client of clients) {
    if (onlyUserId && client.userId !== onlyUserId) continue;
    send(client.socket, {
      type: "comment_likes_update",
      commentId,
      likeCount,
      likerId,
      action,
    });
  }
}

function sendToPlaylistAudience(
  playlist: Pick<Playlist, "isPublic" | "userId">,
  data: ServerToClientMessage,
): void {
  for (const client of clients) {
    if (playlist.isPublic || client.userId === playlist.userId) {
      send(client.socket, data);
    }
  }
}

export function broadcastPlaylistCreated(playlist: Playlist): void {
  sendToPlaylistAudience(playlist, { type: "playlist_created", playlist });
}

// `wasPublic`: the visibility before this change. A playlist that just went
// private still goes to everyone, so non-owners can drop it from their views;
// one that was and stays private goes to its owner alone.
export function broadcastPlaylistUpdated(
  playlist: Playlist,
  wasPublic: boolean,
): void {
  sendToPlaylistAudience(
    { isPublic: playlist.isPublic || wasPublic, userId: playlist.userId },
    { type: "playlist_updated", playlist },
  );
}

export function broadcastPlaylistDeleted(
  playlistId: string,
  userId: string,
  isPublic: boolean,
): void {
  sendToPlaylistAudience({ isPublic, userId }, {
    type: "playlist_deleted",
    playlistId,
    userId,
  });
}

export function broadcastPlaylistDumpsUpdated(
  playlist: Playlist,
  dumpIds: string[],
): void {
  sendToPlaylistAudience(playlist, {
    type: "playlist_dumps_updated",
    playlistId: playlist.id,
    dumpIds,
  });
}

export function broadcastUserUpdated(
  user: Omit<User, "passwordHash" | "email">,
): void {
  for (const client of clients) {
    send(client.socket, { type: "user_updated", user });
  }
}

export function broadcastCommentCreated(comment: Comment): void {
  for (const client of clients) {
    send(client.socket, { type: "comment_created", comment });
  }
}

export function broadcastCommentDeleted(
  commentId: string,
  dumpId: string,
): void {
  for (const client of clients) {
    send(client.socket, { type: "comment_deleted", commentId, dumpId });
  }
}

export function broadcastCommentUpdated(comment: Comment): void {
  for (const client of clients) {
    send(client.socket, { type: "comment_updated", comment });
  }
}

// Global chat room for members: history is behind auth, so the live stream is
// too — an anonymous socket gets nothing chat-related.
function sendToMembers(data: ServerToClientMessage): void {
  for (const client of clients) {
    if (client.userId) send(client.socket, data);
  }
}

// `nonce` is the sender's id for its pending copy; harmless to everyone else.
export function broadcastChatMessage(
  message: ChatMessage,
  nonce?: string,
): void {
  sendToMembers({ type: "chat_message", message, nonce });
}

export function broadcastChatMessageUpdated(message: ChatMessage): void {
  sendToMembers({ type: "chat_message_updated", message });
}

export function broadcastChatMessageDeleted(id: string): void {
  sendToMembers({ type: "chat_message_deleted", id });
}

// Typing indicator: a client's "typing" signal is considered live for this long
// after its last refresh. Must exceed the client's typing-ping cadence so a
// steady typist never flickers out between pings.
const TYPING_TTL = 6_000;

// Deduped set of users currently composing a chat message (typing signal within
// the TTL), shaped like OnlineUser so the client renders the same avatars.
export function getTypingUsers(): OnlineUser[] {
  const now = Date.now();
  const seen = new Map<string, OnlineUser>();
  for (const client of clients) {
    if (
      client.userId && client.typingAt && now - client.typingAt < TYPING_TTL &&
      !seen.has(client.userId)
    ) {
      seen.set(client.userId, {
        userId: client.userId,
        username: client.username!,
        hasAvatar: !!client.avatarMime,
        avatarVersion: client.avatarVersion,
      });
    }
  }
  return Array.from(seen.values());
}

export function broadcastTyping(): void {
  sendToMembers({ type: "chat_typing_update", users: getTypingUsers() });
}

// Record (or clear) a client's typing state and tell everyone. The client also
// resends "true" periodically while composing; the sweep below expires it if
// they go quiet without a clean "false" (e.g. they closed the tab).
export function setClientTyping(client: WsClient, typing: boolean): void {
  const wasTyping = !!client.typingAt;
  client.typingAt = typing ? Date.now() : undefined;
  // Only rebroadcast on an actual edge to avoid a flood of identical updates
  // from the periodic typing pings.
  if (typing !== wasTyping) broadcastTyping();
}

// Sweep stale typing entries (clients that stopped pinging without a clean
// stop) and rebroadcast when the visible set shrinks.
setInterval(() => {
  const now = Date.now();
  let changed = false;
  for (const client of clients) {
    if (client.typingAt && now - client.typingAt >= TYPING_TTL) {
      client.typingAt = undefined;
      changed = true;
    }
  }
  if (changed) broadcastTyping();
}, 2_000);

// True if the user has the chatbox open on any of their connected clients — i.e.
// they're already reading chat, so a mention notification would be redundant.
export function isUserChatOpen(userId: string): boolean {
  for (const client of clients) {
    if (client.userId === userId && client.chatOpen) return true;
  }
  return false;
}

export { type RateLimit, takeToken } from "../lib/rate-limit.ts";

export function handleClientPong(client: WsClient): void {
  client.pongReceived = true;
}

// Keepalive: ping all clients every 30s, disconnect non-responsive ones
const PING_INTERVAL = 30_000;

setInterval(() => {
  for (const client of clients) {
    if (client.socket.readyState !== WebSocket.OPEN) {
      clients.delete(client);
      continue;
    }
    // Disconnect if no pong since last ping (pongReceived starts undefined, skip first cycle)
    if (client.pongReceived === false) {
      client.socket.close(1001, "Ping timeout");
      clients.delete(client);
      continue;
    }
    client.pongReceived = false;
    send(client.socket, { type: "ping" });
  }
}, PING_INTERVAL);