520 lines
14 KiB
TypeScript
520 lines
14 KiB
TypeScript
/**
|
|
* Opens the WebSocket that `WSProvider` keeps for each tab, and handles what
|
|
* the client sends over it. Voting, comment likes and every chat write go
|
|
* through here, not through HTTP. The messages are typed in `shared/ws.ts`,
|
|
* and `services/ws-service.ts` sends the broadcasts.
|
|
*
|
|
* - `GET /ws?token=`, public. A valid session `token` signs the socket in, and
|
|
* an invalid one leaves it anonymous. Answers 403 to an origin not in
|
|
* {@link ALLOWED_ORIGINS}, and 426 to a request that isn't an upgrade.
|
|
*
|
|
* @module
|
|
*/
|
|
|
|
import { Router } from "@oak/oak";
|
|
import { ALLOWED_ORIGINS } from "../config.ts";
|
|
import { authenticate } from "../lib/auth.ts";
|
|
import {
|
|
broadcastChatMessage,
|
|
broadcastChatMessageDeleted,
|
|
broadcastChatMessageUpdated,
|
|
broadcastCommentLikeUpdate,
|
|
broadcastPresence,
|
|
broadcastTyping,
|
|
broadcastVoteUpdate,
|
|
getOnlineUsers,
|
|
handleClientPong,
|
|
isUserChatOpen,
|
|
type RateLimit,
|
|
register,
|
|
send,
|
|
sendToUser,
|
|
setClientTyping,
|
|
takeToken,
|
|
unregister,
|
|
type WsClient,
|
|
} from "../services/ws-service.ts";
|
|
import {
|
|
chatMessageMentionsEveryone,
|
|
createChatMessage,
|
|
deleteChatMessage,
|
|
getChatReadState,
|
|
markChatRead,
|
|
updateChatMessage,
|
|
} from "../services/chat-service.ts";
|
|
import {
|
|
getUnreadCount,
|
|
notifyEveryone,
|
|
notifyMentions,
|
|
} from "../services/notification-service.ts";
|
|
import { hasPermission } from "../../shared/permissions.ts";
|
|
import {
|
|
castVote,
|
|
getUserVotes,
|
|
removeVote,
|
|
} from "../services/vote-service.ts";
|
|
import {
|
|
castCommentLike,
|
|
getUserCommentLikes,
|
|
removeCommentLike,
|
|
} from "../services/comment-like-service.ts";
|
|
import { getUserById } from "../services/user-service.ts";
|
|
import { APIException } from "../lib/api-exception.ts";
|
|
import type { ClientToServerMessage } from "../../shared/ws.ts";
|
|
import type { User } from "../model/user.ts";
|
|
|
|
/**
|
|
* Mounted by `api/main.ts` behind the error, Open Graph and CORS middleware.
|
|
* An `APIException` thrown by a handler becomes the JSON error response there.
|
|
*/
|
|
const router: Router = new Router();
|
|
|
|
/**
|
|
* Checks whether a socket may be opened from `origin`. Browsers don't apply
|
|
* CORS to WebSockets, so this is what stops another site from opening one with
|
|
* its visitors' tokens.
|
|
*/
|
|
function isAllowedOrigin(origin: string): boolean {
|
|
return ALLOWED_ORIGINS.includes(origin);
|
|
}
|
|
|
|
router.get("/ws", async (ctx) => {
|
|
const origin = ctx.request.headers.get("origin") ?? "";
|
|
if (!isAllowedOrigin(origin)) {
|
|
ctx.response.status = 403;
|
|
return;
|
|
}
|
|
|
|
if (!ctx.isUpgradable) {
|
|
ctx.response.status = 426;
|
|
return;
|
|
}
|
|
|
|
// A revoked token (deleted user, changed password) opens an anonymous
|
|
// socket. The client signs out on its next API call, which gets a 401.
|
|
const token = ctx.request.url.searchParams.get("token");
|
|
const authPayload = token ? await authenticate(token) : null;
|
|
let user: User | undefined;
|
|
try {
|
|
user = authPayload ? getUserById(authPayload.userId) : undefined;
|
|
} catch {
|
|
user = undefined;
|
|
}
|
|
|
|
const socket = ctx.upgrade();
|
|
|
|
const client: WsClient = {
|
|
socket,
|
|
userId: user?.id,
|
|
username: user?.username,
|
|
role: user?.role,
|
|
avatarMime: user?.avatarMime,
|
|
};
|
|
|
|
// addEventListener, since `onopen =` proved unreliable under Deno.serve.
|
|
socket.addEventListener("open", () => {
|
|
register(client);
|
|
// The new socket is registered, so it gets this presence_update too,
|
|
// before its welcome.
|
|
broadcastPresence();
|
|
|
|
try {
|
|
const myVotes = user ? getUserVotes(user.id) : [];
|
|
const myCommentLikes = user ? getUserCommentLikes(user.id) : [];
|
|
const unreadNotificationCount = user ? getUnreadCount(user.id) : 0;
|
|
const chatRead = user ? getChatReadState(user.id) : undefined;
|
|
send(socket, {
|
|
type: "welcome",
|
|
users: getOnlineUsers(),
|
|
myVotes,
|
|
myCommentLikes,
|
|
unreadNotificationCount,
|
|
chatReadAt: chatRead?.readAt,
|
|
unreadChatCount: chatRead?.unread ?? 0,
|
|
});
|
|
} catch (err) {
|
|
console.error("[ws] welcome send failed:", err);
|
|
}
|
|
});
|
|
|
|
socket.addEventListener("message", (event) => {
|
|
// An exception escaping a WebSocket listener is uncaught in Deno and ends
|
|
// the process, so nothing a client sends may throw out of here.
|
|
try {
|
|
handleMessage(client, event.data);
|
|
} catch (err) {
|
|
console.error("[ws] message handler failed:", err);
|
|
}
|
|
});
|
|
|
|
socket.addEventListener("close", () => {
|
|
unregister(client);
|
|
broadcastPresence();
|
|
// The sweep only sees registered clients, so it would never clear this one.
|
|
if (client.typingAt) broadcastTyping();
|
|
});
|
|
});
|
|
|
|
/**
|
|
* Parses one message from a client and dispatches it by `type`. A message that
|
|
* isn't JSON, or whose `type` is unknown, is ignored. Its fields are still
|
|
* untrusted, so each handler checks their types.
|
|
*/
|
|
function handleMessage(client: WsClient, data: unknown): void {
|
|
if (typeof data !== "string") return;
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(data);
|
|
} catch {
|
|
return;
|
|
}
|
|
if (
|
|
typeof parsed !== "object" || parsed === null ||
|
|
typeof (parsed as { type?: unknown }).type !== "string"
|
|
) return;
|
|
const msg = parsed as ClientToServerMessage;
|
|
const { socket } = client;
|
|
|
|
switch (msg.type) {
|
|
case "ping":
|
|
send(socket, { type: "pong" });
|
|
break;
|
|
case "pong":
|
|
handleClientPong(client);
|
|
break;
|
|
case "vote_cast":
|
|
handleVote(client, msg.dumpId, "cast");
|
|
break;
|
|
case "vote_remove":
|
|
handleVote(client, msg.dumpId, "remove");
|
|
break;
|
|
case "comment_like_cast":
|
|
handleCommentLike(client, msg.commentId, "cast");
|
|
break;
|
|
case "comment_like_remove":
|
|
handleCommentLike(client, msg.commentId, "remove");
|
|
break;
|
|
case "chat_send":
|
|
handleChatSend(client, msg.body, msg.replyToId, msg.nonce);
|
|
break;
|
|
case "chat_typing":
|
|
// Typing is cosmetic: past the limit it's dropped, with no error.
|
|
if (
|
|
client.userId &&
|
|
takeToken(`typing:${client.userId}`, TYPING_LIMIT)
|
|
) {
|
|
setClientTyping(client, msg.typing === true);
|
|
}
|
|
break;
|
|
case "chat_focus":
|
|
client.chatOpen = msg.open === true;
|
|
break;
|
|
case "chat_edit":
|
|
handleChatEdit(client, msg.id, msg.body);
|
|
break;
|
|
case "chat_delete":
|
|
handleChatDelete(client, msg.id);
|
|
break;
|
|
case "chat_read":
|
|
handleChatRead(client, msg.at);
|
|
break;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handles `vote_cast` and `vote_remove`. It acks the voter with `vote_ack` and
|
|
* broadcasts `votes_update`. A failure goes back as an `error` naming the dump
|
|
* and the action, so the client can undo its optimistic vote at once.
|
|
*/
|
|
function handleVote(
|
|
client: WsClient,
|
|
dumpId: unknown,
|
|
action: "cast" | "remove",
|
|
): void {
|
|
const { socket } = client;
|
|
|
|
if (!client.userId) {
|
|
send(socket, { type: "error", message: "Authentication required" });
|
|
return;
|
|
}
|
|
if (typeof dumpId !== "string") return;
|
|
|
|
try {
|
|
const { voteCount, isPrivate } = action === "cast"
|
|
? castVote(dumpId, client.userId)
|
|
: removeVote(dumpId, client.userId);
|
|
|
|
send(socket, {
|
|
type: "vote_ack",
|
|
dumpId,
|
|
action,
|
|
voteCount,
|
|
});
|
|
|
|
// Only a private dump's owner can see it, and so vote on it.
|
|
broadcastVoteUpdate(
|
|
dumpId,
|
|
voteCount,
|
|
client.userId,
|
|
action,
|
|
isPrivate ? client.userId : undefined,
|
|
);
|
|
} catch (err) {
|
|
const message = err instanceof APIException ? err.message : "Vote failed";
|
|
send(socket, { type: "error", message, dumpId, action });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handles `comment_like_cast` and `comment_like_remove`, as
|
|
* {@link handleVote} does votes.
|
|
*/
|
|
function handleCommentLike(
|
|
client: WsClient,
|
|
commentId: unknown,
|
|
action: "cast" | "remove",
|
|
): void {
|
|
const { socket } = client;
|
|
|
|
if (!client.userId) {
|
|
send(socket, { type: "error", message: "Authentication required" });
|
|
return;
|
|
}
|
|
if (typeof commentId !== "string") return;
|
|
|
|
try {
|
|
const { likeCount, isPrivate } = action === "cast"
|
|
? castCommentLike(commentId, client.userId)
|
|
: removeCommentLike(commentId, client.userId);
|
|
|
|
send(socket, {
|
|
type: "comment_like_ack",
|
|
commentId,
|
|
action,
|
|
likeCount,
|
|
});
|
|
|
|
broadcastCommentLikeUpdate(
|
|
commentId,
|
|
likeCount,
|
|
client.userId,
|
|
action,
|
|
isPrivate ? client.userId : undefined,
|
|
);
|
|
} catch (err) {
|
|
const message = err instanceof APIException ? err.message : "Like failed";
|
|
send(socket, { type: "error", message, commentId, action });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Limits each user's chat writes, sends, edits and deletes together, to a burst
|
|
* of 8, then one every 1.5 s. The buckets are per user, so more tabs don't buy
|
|
* more.
|
|
*/
|
|
const CHAT_WRITE_LIMIT: RateLimit = { capacity: 8, refill: 1 / 1.5 };
|
|
const TYPING_LIMIT: RateLimit = { capacity: 10, refill: 1 };
|
|
/**
|
|
* Limits each user's `chat_read`. One over the limit is dropped, and the next
|
|
* one catches up, since a marker only moves forward.
|
|
*/
|
|
const CHAT_READ_LIMIT: RateLimit = { capacity: 10, refill: 1 };
|
|
|
|
/**
|
|
* Sends a chat-scoped `error`. With the `nonce` of a `chat_send`, the chatbox
|
|
* marks that pending message failed. Without one, it shows `message` as a
|
|
* notice.
|
|
*/
|
|
function sendChatError(
|
|
client: WsClient,
|
|
message: string,
|
|
nonce?: string,
|
|
): void {
|
|
send(client.socket, { type: "error", scope: "chat", message, nonce });
|
|
}
|
|
|
|
/**
|
|
* Returns the id of the user behind a chat write, or undefined after sending
|
|
* the error when they are signed out or over {@link CHAT_WRITE_LIMIT}.
|
|
*/
|
|
function chatWriter(client: WsClient, nonce?: string): string | undefined {
|
|
if (!client.userId) {
|
|
sendChatError(client, "Authentication required", nonce);
|
|
return undefined;
|
|
}
|
|
if (!takeToken(`chat:${client.userId}`, CHAT_WRITE_LIMIT)) {
|
|
sendChatError(
|
|
client,
|
|
"You're sending messages too fast — wait a moment",
|
|
nonce,
|
|
);
|
|
return undefined;
|
|
}
|
|
return client.userId;
|
|
}
|
|
|
|
/**
|
|
* Returns `value` when it is a string of at most `maxLength`, else undefined.
|
|
*/
|
|
function optionalString(value: unknown, maxLength: number): string | undefined {
|
|
return typeof value === "string" && value.length <= maxLength
|
|
? value
|
|
: undefined;
|
|
}
|
|
|
|
/**
|
|
* Handles `chat_send`. It stores the message, broadcasts it, and notifies the
|
|
* @mentioned users who aren't reading the chat, and everyone who isn't when
|
|
* the message is flagged `mentionsEveryone`.
|
|
*/
|
|
function handleChatSend(
|
|
client: WsClient,
|
|
body: unknown,
|
|
replyToId: unknown,
|
|
rawNonce: unknown,
|
|
): void {
|
|
const nonce = optionalString(rawNonce, 64);
|
|
const userId = chatWriter(client, nonce);
|
|
if (!userId) return;
|
|
if (typeof body !== "string") {
|
|
sendChatError(client, "Message cannot be empty", nonce);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const message = createChatMessage(
|
|
userId,
|
|
body,
|
|
optionalString(replyToId, 64),
|
|
);
|
|
// Before the broadcast, so the typing indicator doesn't linger beside the
|
|
// message.
|
|
setClientTyping(client, false);
|
|
broadcastChatMessage(message, nonce);
|
|
notifyMentions(
|
|
userId,
|
|
message.body,
|
|
"chat",
|
|
message.id,
|
|
"",
|
|
undefined,
|
|
isUserChatOpen,
|
|
);
|
|
if (message.mentionsEveryone) {
|
|
notifyEveryone(userId, message.id, isUserChatOpen);
|
|
}
|
|
} catch (err) {
|
|
sendChatError(
|
|
client,
|
|
err instanceof APIException ? err.message : "Failed to send message",
|
|
nonce,
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Checks whether the client's user may edit and delete others' chat messages.
|
|
* The role is read from the database, not from the socket's connect, so that a
|
|
* demoted moderator loses the right at once.
|
|
*/
|
|
function canModerateChat(client: WsClient): boolean {
|
|
if (!client.userId) return false;
|
|
try {
|
|
client.role = getUserById(client.userId).role;
|
|
} catch {
|
|
client.role = undefined;
|
|
}
|
|
return !!client.role && hasPermission(client.role, "chat:moderate");
|
|
}
|
|
|
|
/**
|
|
* Handles `chat_edit`. It stores the new body, broadcasts it, and notifies, in
|
|
* the author's name, the users the edit newly mentions, and everyone when the
|
|
* edit adds a `@everyone` the author may say.
|
|
*/
|
|
function handleChatEdit(client: WsClient, id: unknown, body: unknown): void {
|
|
const userId = chatWriter(client);
|
|
if (!userId) return;
|
|
if (typeof id !== "string" || typeof body !== "string") {
|
|
sendChatError(client, "Invalid message");
|
|
return;
|
|
}
|
|
|
|
try {
|
|
// Only an edit that adds `@everyone` notifies everyone: fixing a typo in
|
|
// an announcement mustn't reach those who had the chat open when it was
|
|
// sent, nor those who joined since.
|
|
const wasEveryone = chatMessageMentionsEveryone(id);
|
|
const message = updateChatMessage(
|
|
id,
|
|
body,
|
|
userId,
|
|
canModerateChat(client),
|
|
);
|
|
broadcastChatMessageUpdated(message);
|
|
// Mentions already notified for this message are deduplicated, so only
|
|
// the new ones notify. A moderator's edit keeps the author's words, so the
|
|
// author mentions.
|
|
notifyMentions(
|
|
message.userId,
|
|
message.body,
|
|
"chat",
|
|
message.id,
|
|
"",
|
|
undefined,
|
|
isUserChatOpen,
|
|
);
|
|
if (message.mentionsEveryone && !wasEveryone) {
|
|
notifyEveryone(message.userId, message.id, isUserChatOpen);
|
|
}
|
|
} catch (err) {
|
|
sendChatError(
|
|
client,
|
|
err instanceof APIException ? err.message : "Failed to edit message",
|
|
);
|
|
}
|
|
}
|
|
|
|
/** Handles `chat_delete`. It deletes the message and broadcasts its id. */
|
|
function handleChatDelete(client: WsClient, id: unknown): void {
|
|
const userId = chatWriter(client);
|
|
if (!userId) return;
|
|
if (typeof id !== "string") {
|
|
sendChatError(client, "Invalid message");
|
|
return;
|
|
}
|
|
|
|
try {
|
|
deleteChatMessage(id, userId, canModerateChat(client));
|
|
broadcastChatMessageDeleted(id);
|
|
} catch (err) {
|
|
sendChatError(
|
|
client,
|
|
err instanceof APIException ? err.message : "Failed to delete message",
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handles `chat_read`. It moves the user's read marker and sends
|
|
* `chat_read_update` to all their sockets, so every tab and device updates its
|
|
* badge together.
|
|
*/
|
|
function handleChatRead(client: WsClient, at: unknown): void {
|
|
if (!client.userId || typeof at !== "string") return;
|
|
if (!takeToken(`read:${client.userId}`, CHAT_READ_LIMIT)) return;
|
|
try {
|
|
const state = markChatRead(client.userId, at);
|
|
sendToUser(client.userId, {
|
|
type: "chat_read_update",
|
|
readAt: state.readAt,
|
|
unread: state.unread,
|
|
});
|
|
} catch {
|
|
// A malformed marker changes nothing, and the client doesn't wait for an
|
|
// answer.
|
|
}
|
|
}
|
|
|
|
export default router;
|