vincent/gerbeur
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v3: added attachments to resources, allow users to paste images into TextEditor, strengthened WS reliability

Browse Source
This commit is contained in:
khannurien
2026-03-27 08:53:32 +00:00
parent ca70bdc14b
commit f0f6472db6
19 changed files with 450 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
src/contexts/WSProvider.tsx
View File

@@ -40,6 +40,12 @@ interface WSProviderProps {
const MAX_BACKOFF = 30_000;
const ACK_TIMEOUT = 5_000;
const CONNECT_TIMEOUT = 2_500;
interface PendingVote {
timeout: ReturnType<typeof setTimeout>;
rollback: () => void;
}
// Minimal runtime check: verify the `type` field is a known string so we can
// safely cast to the discriminated union and let TypeScript narrow from there.
@@ -58,6 +64,10 @@ function parseWSMessage(data: string): IncomingWSMessage | null {
export function WSProvider(
{ children, token, userId, onForceLogout }: WSProviderProps,
) {
const [wsStatus, setWSStatus] = useState<
"connecting" | "connected" | "disconnected"
>("connecting");
const [wsErrorMessage, setWSErrorMessage] = useState<string | null>(null);
const [onlineUsers, setOnlineUsers] = useState<OnlineUser[]>([]);
const [voteCounts, setVoteCounts] = useState<Record<string, number>>({});
const [myVotes, setMyVotes] = useState<Set<string>>(new Set());
@@ -91,15 +101,46 @@ export function WSProvider(
});
const socketRef = useRef<WebSocket | null>(null);
// Tracks pending optimistic votes: dumpId → revert timeout ID
const pendingRef = useRef<Map<string, ReturnType<typeof setTimeout>>>(
// Tracks pending optimistic votes: dumpId → pending rollback handler
const pendingRef = useRef<Map<string, PendingVote>>(
new Map(),
);
const clearPendingVote = useCallback((dumpId: string) => {
const pending = pendingRef.current.get(dumpId);
if (!pending) return;
clearTimeout(pending.timeout);
pendingRef.current.delete(dumpId);
}, []);
const clearAllPendingVotes = useCallback(() => {
for (const pending of pendingRef.current.values()) {
clearTimeout(pending.timeout);
}
pendingRef.current.clear();
}, []);
const schedulePendingVote = useCallback((
dumpId: string,
rollback: () => void,
) => {
clearPendingVote(dumpId);
const timeout = setTimeout(() => {
pendingRef.current.delete(dumpId);
rollback();
}, ACK_TIMEOUT);
pendingRef.current.set(dumpId, { timeout, rollback });
}, [clearPendingVote]);
useEffect(() => {
let closed = false;
let backoff = 500;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
let connectTimeout: ReturnType<typeof setTimeout> | null = null;
let everConnected = false;
setWSStatus("connecting");
setWSErrorMessage(null);
function connect() {
if (closed) return;
@@ -110,6 +151,25 @@ export function WSProvider(
const ws = new WebSocket(url);
socketRef.current = ws;
connectTimeout = setTimeout(() => {
if (ws.readyState !== WebSocket.CONNECTING) return;
setWSStatus("disconnected");
setWSErrorMessage(
"Can't connect to the live updates server. Upvotes and notifications may not sync until it reconnects.",
);
ws.close();
}, CONNECT_TIMEOUT);
ws.onopen = () => {
if (connectTimeout) {
clearTimeout(connectTimeout);
connectTimeout = null;
}
everConnected = true;
setWSStatus("connected");
setWSErrorMessage(null);
};
ws.onmessage = (event) => {
const msg = parseWSMessage(event.data);
if (!msg) return;
@@ -126,6 +186,9 @@ export function WSProvider(
setOnlineUsers(msg.users);
setMyVotes(new Set(msg.myVotes));
setUnreadNotificationCount(msg.unreadNotificationCount);
// welcome provides authoritative server state — cancel any
// in-flight revert timers, they are now superseded.
clearAllPendingVotes();
break;
case "presence_update":
@@ -139,6 +202,7 @@ export function WSProvider(
// Keep myVotes in sync across tabs: if this vote event belongs to
// the current user (from another tab), update myVotes accordingly.
if (voterId === userIdRef.current) {
clearPendingVote(dumpId);
setMyVotes((prev) => {
const next = new Set(prev);
if (action === "cast") next.add(dumpId);
@@ -182,12 +246,7 @@ export function WSProvider(
case "vote_ack": {
const { dumpId, action, voteCount } = msg;
// Clear pending revert timeout
const timeout = pendingRef.current.get(dumpId);
if (timeout !== undefined) {
clearTimeout(timeout);
pendingRef.current.delete(dumpId);
}
clearPendingVote(dumpId);
// Reconcile with authoritative count
setVoteCounts((prev) => ({ ...prev, [dumpId]: voteCount }));
// Confirm vote state
@@ -272,14 +331,24 @@ export function WSProvider(
break;
case "error":
// On error, revert any pending optimistic update for the affected dump
// (the revert timeout will handle it)
// Vote errors currently don't identify which dump/action failed, so
// fall back to the per-dump timeout rollback instead of guessing.
break;
}
};
ws.onclose = () => {
if (connectTimeout) {
clearTimeout(connectTimeout);
connectTimeout = null;
}
if (closed) return;
setWSStatus("disconnected");
setWSErrorMessage(
everConnected
? "Live updates are temporarily disconnected. Trying to reconnect..."
: "Can't connect to the live updates server. Upvotes and notifications may not sync until it reconnects.",
);
reconnectTimer = setTimeout(() => {
backoff = Math.min(backoff * 2, MAX_BACKOFF);
connect();
@@ -297,12 +366,15 @@ export function WSProvider(
return () => {
closed = true;
if (reconnectTimer) clearTimeout(reconnectTimer);
if (connectTimeout) clearTimeout(connectTimeout);
socketRef.current?.close();
socketRef.current = null;
for (const t of pending.values()) clearTimeout(t);
for (const pendingVote of pending.values()) {
clearTimeout(pendingVote.timeout);
}
pending.clear();
};
}, [token]);
}, [clearAllPendingVotes, clearPendingVote, token]);
const castVote = useCallback((dumpId: string) => {
// Optimistic update
@@ -317,22 +389,23 @@ export function WSProvider(
});
setVoteCounts((prev) => ({ ...prev, [dumpId]: prevCount + 1 }));
// Schedule revert if no ack
const timeout = setTimeout(() => {
pendingRef.current.delete(dumpId);
// Schedule revert if no authoritative confirmation arrives.
schedulePendingVote(dumpId, () => {
setMyVotes((prev) => {
const n = new Set(prev);
n.delete(dumpId);
return n;
});
setVoteCounts((prev) => ({ ...prev, [dumpId]: prevCount }));
}, ACK_TIMEOUT);
pendingRef.current.set(dumpId, timeout);
});
socketRef.current?.send(
JSON.stringify({ type: "vote_cast", dumpId } satisfies OutgoingWSMessage),
);
}, []);
if (socketRef.current?.readyState === WebSocket.OPEN) {
socketRef.current.send(
JSON.stringify({ type: "vote_cast", dumpId } satisfies OutgoingWSMessage),
);
}
// If socket is not OPEN, the revert timer will handle cleanup after ACK_TIMEOUT
}, [schedulePendingVote]);
const removeVote = useCallback((dumpId: string) => {
// Optimistic update
@@ -350,24 +423,25 @@ export function WSProvider(
[dumpId]: Math.max(0, prevCount - 1),
}));
// Schedule revert if no ack
const timeout = setTimeout(() => {
pendingRef.current.delete(dumpId);
// Schedule revert if no authoritative confirmation arrives.
schedulePendingVote(dumpId, () => {
setMyVotes((prev) => {
const n = new Set(prev);
n.add(dumpId);
return n;
});
setVoteCounts((prev) => ({ ...prev, [dumpId]: prevCount }));
}, ACK_TIMEOUT);
pendingRef.current.set(dumpId, timeout);
});
socketRef.current?.send(
JSON.stringify(
{ type: "vote_remove", dumpId } satisfies OutgoingWSMessage,
),
);
}, []);
if (socketRef.current?.readyState === WebSocket.OPEN) {
socketRef.current.send(
JSON.stringify(
{ type: "vote_remove", dumpId } satisfies OutgoingWSMessage,
),
);
}
// If socket is not OPEN, the revert timer will handle cleanup after ACK_TIMEOUT
}, [schedulePendingVote]);
const injectDump = useCallback((dump: Dump) => {
setRecentDumps((prev) => {
@@ -381,6 +455,8 @@ export function WSProvider(
}, []);
const value: WSContextValue = useMemo(() => ({
wsStatus,
wsErrorMessage,
onlineUsers,
voteCounts,
myVotes,
@@ -399,6 +475,8 @@ export function WSProvider(
injectDump,
clearUnreadNotifications,
}), [
wsStatus,
wsErrorMessage,
onlineUsers,
voteCounts,
myVotes,