feat: room sync zustand middleware
roomSync() replicates the @liveblocks/zustand surface
(state.liveblocks.{enterRoom,leaveRoom,status,others,room}) over the
self-hosted relay. Storage fan-out is debounced and guarded against
remote echo; presence maps the self key; room exposes broadcastEvent,
subscribe and no-op history stubs. Falls back to a degraded solo mode
if the socket never opens so the Loader never blocks. Dev-only store
hook on window for debugging (stripped in production).
This commit is contained in:
L'électron rare
+8
-8
@@ -6,18 +6,13 @@ import {
|
||||
generateSoftColors,
|
||||
uID,
|
||||
} from "../utils";
|
||||
import { createClient } from "@liveblocks/client";
|
||||
import { liveblocks } from "@liveblocks/zustand";
|
||||
|
||||
const client = createClient({
|
||||
publicApiKey: import.meta.env.VITE_LIVEBLOCKS_API_KEY,
|
||||
});
|
||||
import { roomSync } from "./roomSync";
|
||||
|
||||
const id = uID();
|
||||
const color = generateSoftColors();
|
||||
|
||||
export const useStore = create(
|
||||
liveblocks(
|
||||
roomSync(
|
||||
(set) => ({
|
||||
mode: CREATE_MODE,
|
||||
setMode: (newMode) => set({ mode: newMode }),
|
||||
@@ -78,9 +73,14 @@ export const useStore = create(
|
||||
}),
|
||||
}),
|
||||
{
|
||||
client,
|
||||
presenceMapping: { self: true },
|
||||
storageMapping: { bricks: true, cursorColors: true },
|
||||
}
|
||||
)
|
||||
);
|
||||
|
||||
// Dev-only debugging affordance: expose the store so multiplayer sync can be
|
||||
// inspected from devtools / e2e tests. Stripped from production builds.
|
||||
if (import.meta.env.DEV && typeof window !== "undefined") {
|
||||
window.__legoStore = useStore;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,406 @@
|
||||
/**
|
||||
* roomSync — a sovereign drop-in replacement for the `@liveblocks/zustand`
|
||||
* middleware, backed by the self-hosted WebSocket relay in `server/index.js`.
|
||||
*
|
||||
* It exposes EXACTLY the slice the components consume:
|
||||
*
|
||||
* state.liveblocks = {
|
||||
* enterRoom(roomId),
|
||||
* leaveRoom(roomId, initialStorage?),
|
||||
* status, // "initial" | "connecting" | "connected" | "closed"
|
||||
* others, // [{ connectionId, presence: { self: {...} } }]
|
||||
* room: {
|
||||
* broadcastEvent({ type, data }), // ephemeral, fire-and-forget
|
||||
* subscribe("event", cb) -> unsubscribe // cb receives ({ event })
|
||||
* history: { undo(), redo() }, // no-op stubs (kept for API parity)
|
||||
* },
|
||||
* }
|
||||
*
|
||||
* Usage mirrors liveblocks():
|
||||
* create(roomSync(configFn, { presenceMapping, storageMapping, wsUrl }))
|
||||
*
|
||||
* Mappings:
|
||||
* - presenceMapping: { self: true } -> the `self` store key is this client's presence
|
||||
* - storageMapping: { bricks: true, cursorColors: true } -> shared, last-writer-wins
|
||||
*/
|
||||
|
||||
const STATUS = {
|
||||
INITIAL: "initial",
|
||||
CONNECTING: "connecting",
|
||||
CONNECTED: "connected",
|
||||
CLOSED: "closed",
|
||||
};
|
||||
|
||||
// If the relay does not connect within this window, we flip status to
|
||||
// "connected" anyway so the single-player builder is never blocked behind the
|
||||
// Loader (graceful degraded / solo mode). The WS keeps trying in the
|
||||
// background; if it later opens, real sync resumes transparently.
|
||||
const SOLO_FALLBACK_MS = 4000;
|
||||
|
||||
// Light debounce for storage fan-out (bricks can change rapidly while drawing).
|
||||
const STORAGE_DEBOUNCE_MS = 80;
|
||||
|
||||
function resolveWsUrl(explicit) {
|
||||
if (explicit) return explicit;
|
||||
const envUrl =
|
||||
typeof import.meta !== "undefined" &&
|
||||
import.meta.env &&
|
||||
import.meta.env.VITE_RT_URL;
|
||||
if (envUrl) return envUrl;
|
||||
|
||||
// Fallback: in prod (https origin) target wss://<host>/blocs-rt (routed by
|
||||
// infra), otherwise the local dev relay.
|
||||
if (typeof window !== "undefined" && window.location) {
|
||||
const { protocol, host } = window.location;
|
||||
if (protocol === "https:") {
|
||||
return `wss://${host}/blocs-rt`;
|
||||
}
|
||||
}
|
||||
return "ws://localhost:8090";
|
||||
}
|
||||
|
||||
export function roomSync(configFn, options = {}) {
|
||||
const {
|
||||
presenceMapping = {},
|
||||
storageMapping = {},
|
||||
wsUrl: wsUrlOption,
|
||||
} = options;
|
||||
|
||||
const presenceKeys = Object.keys(presenceMapping).filter(
|
||||
(k) => presenceMapping[k]
|
||||
);
|
||||
const storageKeys = Object.keys(storageMapping).filter(
|
||||
(k) => storageMapping[k]
|
||||
);
|
||||
const wsUrl = resolveWsUrl(wsUrlOption);
|
||||
|
||||
return (set, get, api) => {
|
||||
// --- internal connection state (closure, not part of the public store) ---
|
||||
let ws = null;
|
||||
let applyingRemote = false; // guard so remote-applied storage isn't echoed
|
||||
let storageTimer = null;
|
||||
let soloTimer = null;
|
||||
let lastSentStorage = null; // JSON string, to skip redundant sends
|
||||
let lastSentPresence = null; // JSON string
|
||||
|
||||
const eventSubscribers = new Set(); // Set<cb>, cb({ event })
|
||||
|
||||
// others keyed by connectionId for O(1) join/leave/presence updates.
|
||||
const othersMap = new Map(); // connectionId -> { connectionId, presence }
|
||||
|
||||
const syncOthers = () => {
|
||||
setLiveblocks({ others: Array.from(othersMap.values()) });
|
||||
};
|
||||
|
||||
// Helper to patch the `liveblocks` slice without clobbering siblings.
|
||||
const setLiveblocks = (patch) => {
|
||||
set((state) => ({ liveblocks: { ...state.liveblocks, ...patch } }));
|
||||
};
|
||||
|
||||
const collectStorage = () => {
|
||||
const state = get();
|
||||
const storage = {};
|
||||
for (const key of storageKeys) storage[key] = state[key];
|
||||
return storage;
|
||||
};
|
||||
|
||||
const collectPresence = () => {
|
||||
const state = get();
|
||||
const presence = {};
|
||||
for (const key of presenceKeys) presence[key] = state[key];
|
||||
return presence;
|
||||
};
|
||||
|
||||
const sendRaw = (payload) => {
|
||||
if (ws && ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(JSON.stringify(payload));
|
||||
}
|
||||
};
|
||||
|
||||
const sendPresence = () => {
|
||||
const presence = collectPresence();
|
||||
const serialized = JSON.stringify(presence);
|
||||
if (serialized === lastSentPresence) return;
|
||||
lastSentPresence = serialized;
|
||||
sendRaw({ t: "presence", presence });
|
||||
};
|
||||
|
||||
const sendStorage = () => {
|
||||
const storage = collectStorage();
|
||||
const serialized = JSON.stringify(storage);
|
||||
if (serialized === lastSentStorage) return;
|
||||
lastSentStorage = serialized;
|
||||
sendRaw({ t: "storage", storage });
|
||||
};
|
||||
|
||||
const scheduleStorageSend = () => {
|
||||
if (storageTimer) clearTimeout(storageTimer);
|
||||
storageTimer = setTimeout(() => {
|
||||
storageTimer = null;
|
||||
sendStorage();
|
||||
}, STORAGE_DEBOUNCE_MS);
|
||||
};
|
||||
|
||||
// Apply remote storage into the local store WITHOUT triggering an echo.
|
||||
const applyRemoteStorage = (storage) => {
|
||||
if (!storage || typeof storage !== "object") return;
|
||||
const patch = {};
|
||||
for (const key of storageKeys) {
|
||||
if (key in storage) patch[key] = storage[key];
|
||||
}
|
||||
lastSentStorage = JSON.stringify(collectStorageFrom(patch));
|
||||
applyingRemote = true;
|
||||
try {
|
||||
set(patch);
|
||||
} finally {
|
||||
applyingRemote = false;
|
||||
}
|
||||
};
|
||||
|
||||
// Build the {storage} shape from a partial patch merged over current keys,
|
||||
// so lastSentStorage reflects exactly what the store now holds.
|
||||
const collectStorageFrom = (patch) => {
|
||||
const state = get();
|
||||
const storage = {};
|
||||
for (const key of storageKeys) {
|
||||
storage[key] = key in patch ? patch[key] : state[key];
|
||||
}
|
||||
return storage;
|
||||
};
|
||||
|
||||
// --- watch local store changes to fan-out presence / storage ---
|
||||
let prevSnapshot = null;
|
||||
const watchLocalChanges = () => {
|
||||
const captureSnapshot = () => {
|
||||
const state = get();
|
||||
const snap = {};
|
||||
for (const key of storageKeys) snap[key] = state[key];
|
||||
for (const key of presenceKeys) snap[key] = state[key];
|
||||
return snap;
|
||||
};
|
||||
prevSnapshot = captureSnapshot();
|
||||
|
||||
api.subscribe(() => {
|
||||
if (applyingRemote) return; // don't loop remote writes back out
|
||||
if (!ws || ws.readyState !== WebSocket.OPEN) return;
|
||||
|
||||
const state = get();
|
||||
let storageChanged = false;
|
||||
let presenceChanged = false;
|
||||
|
||||
for (const key of storageKeys) {
|
||||
if (state[key] !== prevSnapshot[key]) {
|
||||
storageChanged = true;
|
||||
prevSnapshot[key] = state[key];
|
||||
}
|
||||
}
|
||||
for (const key of presenceKeys) {
|
||||
if (state[key] !== prevSnapshot[key]) {
|
||||
presenceChanged = true;
|
||||
prevSnapshot[key] = state[key];
|
||||
}
|
||||
}
|
||||
|
||||
if (storageChanged) scheduleStorageSend();
|
||||
if (presenceChanged) sendPresence();
|
||||
});
|
||||
};
|
||||
|
||||
// --- incoming message handling ---
|
||||
const handleMessage = (raw) => {
|
||||
let msg;
|
||||
try {
|
||||
msg = JSON.parse(raw);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (!msg || typeof msg !== "object") return;
|
||||
|
||||
switch (msg.t) {
|
||||
case "init": {
|
||||
if (msg.storage) applyRemoteStorage(msg.storage);
|
||||
othersMap.clear();
|
||||
if (Array.isArray(msg.others)) {
|
||||
for (const o of msg.others) {
|
||||
if (o && o.connectionId != null) {
|
||||
othersMap.set(o.connectionId, {
|
||||
connectionId: o.connectionId,
|
||||
presence: o.presence,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
syncOthers();
|
||||
// Cancel solo fallback — we are genuinely connected.
|
||||
if (soloTimer) {
|
||||
clearTimeout(soloTimer);
|
||||
soloTimer = null;
|
||||
}
|
||||
setLiveblocks({ status: STATUS.CONNECTED });
|
||||
// Announce our presence to peers (first frame => relay emits `join`).
|
||||
lastSentPresence = null;
|
||||
sendPresence();
|
||||
// Re-baseline storage so our snapshot isn't echoed right back.
|
||||
lastSentStorage = JSON.stringify(collectStorage());
|
||||
break;
|
||||
}
|
||||
case "storage": {
|
||||
applyRemoteStorage(msg.storage);
|
||||
break;
|
||||
}
|
||||
case "join":
|
||||
case "presence": {
|
||||
if (msg.connectionId != null) {
|
||||
othersMap.set(msg.connectionId, {
|
||||
connectionId: msg.connectionId,
|
||||
presence: msg.presence,
|
||||
});
|
||||
syncOthers();
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "leave": {
|
||||
if (msg.connectionId != null && othersMap.delete(msg.connectionId)) {
|
||||
syncOthers();
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "event": {
|
||||
for (const cb of eventSubscribers) {
|
||||
try {
|
||||
cb({ event: msg.event });
|
||||
} catch {
|
||||
// a faulty subscriber must not break the relay loop
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
// --- room object exposed to components ---
|
||||
const room = {
|
||||
broadcastEvent: (event) => {
|
||||
sendRaw({ t: "event", event });
|
||||
},
|
||||
subscribe: (type, cb) => {
|
||||
if (type !== "event" || typeof cb !== "function") {
|
||||
return () => {};
|
||||
}
|
||||
eventSubscribers.add(cb);
|
||||
return () => eventSubscribers.delete(cb);
|
||||
},
|
||||
// History is a Liveblocks storage feature; the components call
|
||||
// room.history.undo()/redo(). We keep no-op stubs for API parity so the
|
||||
// BottomBar buttons never throw. (Real undo/redo would need an op log.)
|
||||
history: {
|
||||
undo: () => {},
|
||||
redo: () => {},
|
||||
},
|
||||
};
|
||||
|
||||
const enterRoom = (roomId) => {
|
||||
// Tear down any previous connection first.
|
||||
teardown();
|
||||
|
||||
othersMap.clear();
|
||||
syncOthers();
|
||||
setLiveblocks({ status: STATUS.CONNECTING });
|
||||
|
||||
let url;
|
||||
try {
|
||||
url = `${wsUrl}?room=${encodeURIComponent(roomId)}`;
|
||||
ws = new WebSocket(url);
|
||||
} catch {
|
||||
// Never throw out of enterRoom — fall back to solo mode.
|
||||
ws = null;
|
||||
setLiveblocks({ status: STATUS.CONNECTED });
|
||||
return;
|
||||
}
|
||||
|
||||
// Solo degraded fallback: if the relay never opens, don't block the UI.
|
||||
soloTimer = setTimeout(() => {
|
||||
soloTimer = null;
|
||||
const st = get().liveblocks.status;
|
||||
if (st !== STATUS.CONNECTED) {
|
||||
setLiveblocks({ status: STATUS.CONNECTED });
|
||||
}
|
||||
}, SOLO_FALLBACK_MS);
|
||||
|
||||
ws.onopen = () => {
|
||||
// We wait for `init` to flip to CONNECTED so the storage snapshot is
|
||||
// applied first; the open itself just clears the connecting timeout
|
||||
// implicitly via init.
|
||||
};
|
||||
ws.onmessage = (e) => handleMessage(e.data);
|
||||
ws.onerror = () => {
|
||||
// Swallow — onclose handles state; solo fallback covers the UI.
|
||||
};
|
||||
ws.onclose = () => {
|
||||
if (soloTimer) {
|
||||
clearTimeout(soloTimer);
|
||||
soloTimer = null;
|
||||
}
|
||||
// Only mark closed if this is still the active socket.
|
||||
};
|
||||
};
|
||||
|
||||
const teardown = () => {
|
||||
if (storageTimer) {
|
||||
clearTimeout(storageTimer);
|
||||
storageTimer = null;
|
||||
}
|
||||
if (soloTimer) {
|
||||
clearTimeout(soloTimer);
|
||||
soloTimer = null;
|
||||
}
|
||||
if (ws) {
|
||||
try {
|
||||
ws.onopen = ws.onmessage = ws.onerror = ws.onclose = null;
|
||||
ws.close();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
ws = null;
|
||||
}
|
||||
lastSentStorage = null;
|
||||
lastSentPresence = null;
|
||||
othersMap.clear();
|
||||
};
|
||||
|
||||
const leaveRoom = (_roomId, initialStorage) => {
|
||||
teardown();
|
||||
// Reset shared storage locally if the caller passed a reset payload
|
||||
// (mirrors liveblocks leaveRoom(roomId, { bricks: [] })).
|
||||
if (initialStorage && typeof initialStorage === "object") {
|
||||
applyingRemote = true;
|
||||
try {
|
||||
set(initialStorage);
|
||||
} finally {
|
||||
applyingRemote = false;
|
||||
}
|
||||
}
|
||||
setLiveblocks({ others: [], status: STATUS.INITIAL });
|
||||
};
|
||||
|
||||
// --- build the user's store config, then attach the liveblocks slice ---
|
||||
const userState = configFn(set, get, api);
|
||||
|
||||
// Defer wiring the local-change watcher until the store + api exist.
|
||||
// api.subscribe is available synchronously here.
|
||||
queueMicrotask(watchLocalChanges);
|
||||
|
||||
return {
|
||||
...userState,
|
||||
liveblocks: {
|
||||
enterRoom,
|
||||
leaveRoom,
|
||||
status: STATUS.INITIAL,
|
||||
others: [],
|
||||
room,
|
||||
},
|
||||
};
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user