353 lines
14 KiB
JavaScript
353 lines
14 KiB
JavaScript
/*
|
|
* Copyright (C) 2026 Mateusz Gruszczyński @linuxiarz.pl
|
|
* Source-Available Code / Dual-Licensed.
|
|
*
|
|
* Free for non-commercial and evaluation use under terms of BSL/GPLv3.
|
|
* Commercial or production use requires a valid paid license.
|
|
* See LICENSE file in repository root for details.
|
|
*/
|
|
|
|
import { logError, logInfo, logWarn } from "@rustpad/logger";
|
|
|
|
const DEFAULT_HEARTBEAT_INTERVAL_MS = 10000;
|
|
const DEFAULT_HEARTBEAT_TIMEOUT_MS = 30000;
|
|
const DEFAULT_MAX_RECONNECT_DELAY_MS = 12000;
|
|
const DEFAULT_LATENCY_SAMPLE_WINDOW = 20;
|
|
|
|
class RoomSocket {
|
|
constructor(options) {
|
|
Object.assign(this, options);
|
|
this.socket = null;
|
|
this.reconnectTimer = null;
|
|
this.pingTimer = null;
|
|
this.stopped = false;
|
|
this.intentionalClose = false;
|
|
this.reconnectAttempt = 0;
|
|
this.pendingPings = new Map();
|
|
this.heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS;
|
|
this.heartbeatTimeoutMs = DEFAULT_HEARTBEAT_TIMEOUT_MS;
|
|
this.maxReconnectDelayMs = DEFAULT_MAX_RECONNECT_DELAY_MS;
|
|
this.latencySampleWindow = DEFAULT_LATENCY_SAMPLE_WINDOW;
|
|
this.latencySamples = [];
|
|
this.totalReconnects = 0;
|
|
this.authenticatedAt = null;
|
|
this.lastConnectionUptimeMs = 0;
|
|
this.lastMessageAt = null;
|
|
this.lastClose = null;
|
|
this.bytesSent = 0;
|
|
this.bytesReceived = 0;
|
|
this.serverDiagnostics = null;
|
|
this.diagnosticsTimer = null;
|
|
this.handleOnline = () => this.reconnectNow("Network connection restored.");
|
|
this.handleOffline = () => this.handleNetworkOffline();
|
|
this.handleVisibility = () => this.checkHeartbeat();
|
|
window.addEventListener("online", this.handleOnline);
|
|
window.addEventListener("offline", this.handleOffline);
|
|
document.addEventListener("visibilitychange", this.handleVisibility);
|
|
}
|
|
|
|
get url() { throw new Error("Socket URL not implemented"); }
|
|
get kind() { return "room"; }
|
|
|
|
clientDiagnostics() {
|
|
return {
|
|
language: navigator.language || null,
|
|
timezone: Intl.DateTimeFormat().resolvedOptions().timeZone || null,
|
|
platform: navigator.userAgentData?.platform || navigator.platform || null,
|
|
};
|
|
}
|
|
|
|
connect() {
|
|
clearTimeout(this.reconnectTimer);
|
|
clearInterval(this.pingTimer);
|
|
clearInterval(this.diagnosticsTimer);
|
|
if (this.stopped) return;
|
|
if (!navigator.onLine) {
|
|
this.scheduleReconnect("Your device is offline.");
|
|
return;
|
|
}
|
|
|
|
this.intentionalClose = false;
|
|
this.onStatus?.(this.reconnectAttempt ? "reconnecting" : "connecting", {
|
|
attempt: this.reconnectAttempt,
|
|
message: this.reconnectAttempt ? "Re-establishing the live connection." : "Opening the live connection.",
|
|
});
|
|
this.emitDiagnostics();
|
|
|
|
let socket;
|
|
try {
|
|
socket = new WebSocket(this.url);
|
|
} catch (error) {
|
|
logError("websocket.create", error, { kind: this.kind });
|
|
this.scheduleReconnect("The live connection could not be opened.");
|
|
return;
|
|
}
|
|
this.socket = socket;
|
|
|
|
socket.addEventListener("open", () => {
|
|
if (socket !== this.socket || this.stopped) return;
|
|
logInfo("websocket.open", { kind: this.kind });
|
|
this.serverDiagnostics = null;
|
|
this.lastClose = null;
|
|
this.send({
|
|
type: "authenticate",
|
|
password: this.password || null,
|
|
access_token: this.accessToken || null,
|
|
nickname: this.nickname || null,
|
|
guest_id: this.guestId || null,
|
|
color: this.color || null,
|
|
diagnostics: this.clientDiagnostics(),
|
|
client_id: this.clientId || null,
|
|
known_revision_id: this.getKnownRevision?.() ?? null,
|
|
});
|
|
this.emitDiagnostics();
|
|
});
|
|
|
|
socket.addEventListener("message", event => {
|
|
if (socket !== this.socket || this.stopped) return;
|
|
let message;
|
|
try { message = JSON.parse(event.data); } catch { return; }
|
|
this.lastMessageAt = Date.now();
|
|
this.bytesReceived += typeof event.data === "string" ? new Blob([event.data]).size : Number(event.data?.byteLength || 0);
|
|
if (message.type === "password_required") {
|
|
this.intentionalClose = true;
|
|
this.onPasswordRequired?.();
|
|
socket.close();
|
|
return;
|
|
}
|
|
if (message.type === "password_changed") {
|
|
this.intentionalClose = true;
|
|
socket.close();
|
|
return;
|
|
}
|
|
if (message.type === "error") {
|
|
this.intentionalClose = true;
|
|
this.onError?.(message.message);
|
|
socket.close();
|
|
return;
|
|
}
|
|
if (message.type === "authenticated") {
|
|
const restored = this.reconnectAttempt > 0;
|
|
this.reconnectAttempt = 0;
|
|
this.authenticatedAt = Date.now();
|
|
this.latencySamples = [];
|
|
this.onStatus?.("online", { restored });
|
|
this.onAuthenticated?.(message);
|
|
this.startPing();
|
|
this.startDiagnostics();
|
|
this.emitDiagnostics();
|
|
return;
|
|
}
|
|
if (message.type === "diagnostics") {
|
|
this.applyServerDiagnostics(message.diagnostics || {});
|
|
return;
|
|
}
|
|
if (message.type === "document") this.onDocument?.(message);
|
|
if (message.type === "resync") this.onResync?.(message);
|
|
if (message.type === "presence") this.onPresence?.(message.users || []);
|
|
if (message.type === "chat") this.onChat?.(message);
|
|
if (message.type === "pong") {
|
|
const started = this.pendingPings.get(message.nonce);
|
|
if (started !== undefined) {
|
|
this.pendingPings.delete(message.nonce);
|
|
const latency = Math.max(0, Math.round(performance.now() - started));
|
|
this.latencySamples.push(latency);
|
|
while (this.latencySamples.length > this.latencySampleWindow) this.latencySamples.shift();
|
|
this.onLatency?.(latency);
|
|
this.emitDiagnostics();
|
|
}
|
|
}
|
|
});
|
|
|
|
socket.addEventListener("close", event => {
|
|
if (socket !== this.socket) return;
|
|
clearInterval(this.pingTimer);
|
|
clearInterval(this.diagnosticsTimer);
|
|
this.pendingPings.clear();
|
|
this.onPresence?.([]);
|
|
this.onLatency?.(null);
|
|
const closedAt = Date.now();
|
|
this.lastConnectionUptimeMs = this.authenticatedAt ? Math.max(0, closedAt - this.authenticatedAt) : this.lastConnectionUptimeMs;
|
|
this.authenticatedAt = null;
|
|
this.lastClose = { code: event.code, reason: event.reason || "", at: closedAt };
|
|
this.emitDiagnostics();
|
|
logWarn("websocket.close", {
|
|
kind: this.kind,
|
|
code: event.code,
|
|
reason: event.reason || "",
|
|
intentional: this.intentionalClose || this.stopped,
|
|
});
|
|
if (!this.intentionalClose && !this.stopped) {
|
|
this.scheduleReconnect(this.closeMessage(event));
|
|
}
|
|
});
|
|
|
|
socket.addEventListener("error", event => {
|
|
if (socket !== this.socket || this.stopped) return;
|
|
logError("websocket.error", event, { kind: this.kind });
|
|
if (socket.readyState !== WebSocket.CLOSING && socket.readyState !== WebSocket.CLOSED) {
|
|
socket.close();
|
|
}
|
|
});
|
|
}
|
|
|
|
closeMessage(event) {
|
|
if (!navigator.onLine) return "Your device is offline.";
|
|
if (event.reason === "heartbeat timeout") return "The connection stopped responding after the tab was inactive.";
|
|
return "The server connection was interrupted.";
|
|
}
|
|
|
|
scheduleReconnect(message) {
|
|
if (this.stopped) return;
|
|
clearTimeout(this.reconnectTimer);
|
|
this.reconnectAttempt += 1;
|
|
const baseDelay = Math.min(this.maxReconnectDelayMs, 750 * (2 ** Math.min(this.reconnectAttempt - 1, 4)));
|
|
const retryInMs = navigator.onLine ? baseDelay : 3000;
|
|
this.totalReconnects += 1;
|
|
this.onStatus?.("reconnecting", { attempt: this.reconnectAttempt, message, retryInMs });
|
|
this.emitDiagnostics();
|
|
this.reconnectTimer = window.setTimeout(() => this.connect(), retryInMs);
|
|
}
|
|
|
|
reconnectNow(message) {
|
|
if (this.stopped || this.socket?.readyState === WebSocket.OPEN || this.socket?.readyState === WebSocket.CONNECTING) return;
|
|
clearTimeout(this.reconnectTimer);
|
|
this.onStatus?.("reconnecting", { attempt: this.reconnectAttempt, message, retryInMs: 0 });
|
|
this.connect();
|
|
}
|
|
|
|
handleNetworkOffline() {
|
|
if (this.stopped) return;
|
|
this.onStatus?.("reconnecting", { attempt: this.reconnectAttempt + 1, message: "Your device is offline.", retryInMs: 3000 });
|
|
if (this.socket?.readyState === WebSocket.OPEN || this.socket?.readyState === WebSocket.CONNECTING) {
|
|
this.socket.close();
|
|
} else {
|
|
this.scheduleReconnect("Your device is offline.");
|
|
}
|
|
}
|
|
|
|
checkHeartbeat() {
|
|
if (this.stopped || this.socket?.readyState !== WebSocket.OPEN) return;
|
|
const now = performance.now();
|
|
const expired = [...this.pendingPings.values()].some(started => now - started >= this.heartbeatTimeoutMs);
|
|
if (expired) {
|
|
logWarn("websocket.heartbeat_timeout", { kind: this.kind });
|
|
this.socket.close(4000, "heartbeat timeout");
|
|
}
|
|
}
|
|
|
|
startPing() {
|
|
clearInterval(this.pingTimer);
|
|
const ping = () => {
|
|
this.checkHeartbeat();
|
|
if (this.socket?.readyState !== WebSocket.OPEN) return;
|
|
const nonce = Date.now();
|
|
this.pendingPings.set(nonce, performance.now());
|
|
this.send({ type: "ping", nonce });
|
|
};
|
|
ping();
|
|
this.pingTimer = window.setInterval(ping, this.heartbeatIntervalMs);
|
|
}
|
|
|
|
startDiagnostics() {
|
|
clearInterval(this.diagnosticsTimer);
|
|
this.diagnosticsTimer = window.setInterval(() => this.emitDiagnostics(), 1000);
|
|
}
|
|
|
|
applyServerDiagnostics(diagnostics) {
|
|
this.serverDiagnostics = diagnostics;
|
|
const positiveNumber = (value, fallback, min, max) => {
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) && parsed >= min && parsed <= max ? parsed : fallback;
|
|
};
|
|
this.heartbeatIntervalMs = positiveNumber(diagnostics.heartbeat_interval_ms, this.heartbeatIntervalMs, 1000, 120000);
|
|
this.heartbeatTimeoutMs = positiveNumber(diagnostics.heartbeat_timeout_ms, this.heartbeatTimeoutMs, this.heartbeatIntervalMs * 2, 300000);
|
|
this.maxReconnectDelayMs = positiveNumber(diagnostics.max_reconnect_delay_ms, this.maxReconnectDelayMs, 1000, 120000);
|
|
this.latencySampleWindow = Math.round(positiveNumber(diagnostics.latency_sample_window, this.latencySampleWindow, 3, 100));
|
|
while (this.latencySamples.length > this.latencySampleWindow) this.latencySamples.shift();
|
|
if (this.socket?.readyState === WebSocket.OPEN && this.authenticatedAt) this.startPing();
|
|
this.emitDiagnostics();
|
|
}
|
|
|
|
latencyStats() {
|
|
if (!this.latencySamples.length) return { current: null, average: null, minimum: null, maximum: null, jitter: null, quality: "unknown" };
|
|
const samples = this.latencySamples;
|
|
const current = samples.at(-1);
|
|
const average = Math.round(samples.reduce((sum, value) => sum + value, 0) / samples.length);
|
|
const minimum = Math.min(...samples);
|
|
const maximum = Math.max(...samples);
|
|
const differences = samples.slice(1).map((value, index) => Math.abs(value - samples[index]));
|
|
const jitter = differences.length ? Math.round(differences.reduce((sum, value) => sum + value, 0) / differences.length) : 0;
|
|
const thresholds = this.serverDiagnostics?.quality_thresholds || {};
|
|
const excellent = Number(thresholds.excellent_max_ms ?? 100);
|
|
const good = Number(thresholds.good_max_ms ?? 250);
|
|
const degraded = Number(thresholds.degraded_max_ms ?? 600);
|
|
const quality = current <= excellent ? "excellent" : current <= good ? "good" : current <= degraded ? "degraded" : "poor";
|
|
return { current, average, minimum, maximum, jitter, quality };
|
|
}
|
|
|
|
emitDiagnostics() {
|
|
const readyStates = ["connecting", "open", "closing", "closed"];
|
|
this.onDiagnostics?.({
|
|
server: this.serverDiagnostics,
|
|
runtime: {
|
|
state: readyStates[this.socket?.readyState ?? WebSocket.CLOSED] || "closed",
|
|
visibility: document.visibilityState,
|
|
uptime_ms: this.authenticatedAt ? Math.max(0, Date.now() - this.authenticatedAt) : 0,
|
|
last_connection_uptime_ms: this.lastConnectionUptimeMs,
|
|
reconnect_attempt: this.reconnectAttempt,
|
|
total_reconnects: this.totalReconnects,
|
|
authenticated_at: this.authenticatedAt,
|
|
last_message_at: this.lastMessageAt,
|
|
last_close: this.lastClose,
|
|
buffered_amount: this.socket?.bufferedAmount || 0,
|
|
bytes_sent: this.bytesSent,
|
|
bytes_received: this.bytesReceived,
|
|
latency: this.latencyStats(),
|
|
},
|
|
});
|
|
}
|
|
|
|
send(message) {
|
|
if (this.socket?.readyState !== WebSocket.OPEN) return false;
|
|
const payload = JSON.stringify(message);
|
|
this.bytesSent += new Blob([payload]).size;
|
|
this.socket.send(payload);
|
|
return true;
|
|
}
|
|
|
|
update(baseRevisionId, updateId, operation, ownerReplacements = []) {
|
|
return this.send({
|
|
type: "update",
|
|
base_revision_id: baseRevisionId,
|
|
update_id: updateId,
|
|
operation,
|
|
owner_replacements: ownerReplacements,
|
|
});
|
|
}
|
|
chat(text) { this.send({ type: "chat", text }); }
|
|
setColor(color) { this.color = color || null; this.send({ type: "set_color", color: this.color }); }
|
|
|
|
stop() {
|
|
this.stopped = true;
|
|
this.intentionalClose = true;
|
|
clearTimeout(this.reconnectTimer);
|
|
clearInterval(this.pingTimer);
|
|
clearInterval(this.diagnosticsTimer);
|
|
window.removeEventListener("online", this.handleOnline);
|
|
window.removeEventListener("offline", this.handleOffline);
|
|
document.removeEventListener("visibilitychange", this.handleVisibility);
|
|
this.socket?.close();
|
|
}
|
|
}
|
|
|
|
export class NoteSocket extends RoomSocket {
|
|
get kind() { return "note"; }
|
|
get url() { const p = location.protocol === "https:" ? "wss:" : "ws:"; return `${p}//${location.host}/ws/${encodeURIComponent(this.workspaceSlug)}/${encodeURIComponent(this.noteSlug)}`; }
|
|
}
|
|
|
|
export class PadSocket extends RoomSocket {
|
|
get kind() { return "pad"; }
|
|
get url() { const p = location.protocol === "https:" ? "wss:" : "ws:"; return `${p}//${location.host}/ws/p/${encodeURIComponent(this.slug)}`; }
|
|
}
|