import { draftPersistence } from "./draft-persistence"; import { createNanoEvents } from "nanoevents"; import { $collaborators, $syncStatus, createMultiplayerRetryTracker, type ApplyMessage, type ErrorMessage, type PresenceMessage, type ReloadMessage, type SyncEmitter, type SyncMessage, } from "@webstudio-is/sync-client"; import { createWebSocketSyncEmitter, type WebSocketEmitterOptions, type WebSocketSyncEmitter, } from "@webstudio-is/sync-client/websocket"; import { $awareness, type Awareness } from "~/shared/awareness"; type MultiplayerSyncEmitterOptions = { projectId?: string; clientId: string; getAuthToken?: () => Promise; url: string; onReload?: (message: ReloadMessage) => void; onUserMessage?: (message: string) => void; createTransport?: (options: WebSocketEmitterOptions) => WebSocketSyncEmitter; }; export type MultiplayerSyncEmitter = SyncEmitter & { close(): void; connect(buildId: string): void; sendPresence(payload: Awareness): void; }; const updateCollaborator = (clientId: string, payload: unknown) => { const next = new Map($collaborators.get()); const previous = next.get(clientId); next.set(clientId, { ...previous, ...(typeof payload === "object" && payload !== null ? payload : {}), lastSeen: Date.now(), }); $collaborators.set(next); }; export const startMultiplayerPresenceSync = ( emitter: Pick ) => { return $awareness.listen((awareness) => { emitter.sendPresence(awareness); }); }; export const createMultiplayerSyncEmitter = ({ projectId, clientId, getAuthToken, url, onReload, onUserMessage, createTransport = createWebSocketSyncEmitter, }: MultiplayerSyncEmitterOptions): MultiplayerSyncEmitter => { const events = createNanoEvents<{ message: (message: SyncMessage) => void; }>(); let clientSeq = 0; let durableSeq = 0; const remoteTransactions = new Map(); let tracker: ReturnType; let ws: WebSocketSyncEmitter | undefined; const updateUnsaved = () => { $syncStatus.set( tracker.getPendingCount() > 0 ? { status: "syncing" } : { status: "idle" } ); }; const sendApply = (message: ApplyMessage) => { if (ws === undefined) { return; } ws.sendToRelay(message); }; const connect = (buildId: string) => { if (ws !== undefined) { return; } ws = createTransport({ buildId, clientId, getAuthToken, url, onAck: (seq, version) => { durableSeq = Math.max(durableSeq, seq); for (const [id, transactionSeq] of remoteTransactions) { if (transactionSeq <= durableSeq) { if (projectId) { draftPersistence.complete(projectId, id, true); } remoteTransactions.delete(id); } } tracker.handleAck({ type: "ack", seq, version }); updateUnsaved(); }, onApplied: (transactionId, seq, status, errors) => { tracker.handleApplied({ type: "applied", transactionId, seq, status, errors, }); updateUnsaved(); }, onBroadcast: (message) => { // Other collaborators' changes become visible before the database ACK too. if ( projectId && message.originClientId !== clientId && message.seq > durableSeq ) { draftPersistence.begin(projectId, message.transaction.id); remoteTransactions.set(message.transaction.id, message.seq); } tracker.handleBroadcast(message); updateUnsaved(); events.emit("message", { type: "apply", clientId: message.originClientId, transaction: message.transaction, }); }, onClose: () => {}, onErrorMessage: (message: ErrorMessage) => { tracker.handleErrorMessage(message); updateUnsaved(); }, onOpen: () => { tracker.handleReconnect(); }, onPresence: (presenceClientId, payload) => { updateCollaborator(presenceClientId, payload); }, onReload: (message) => { tracker.handleReload(message); updateUnsaved(); }, }); }; tracker = createMultiplayerRetryTracker({ onDropped: ({ message }) => { if (projectId) { draftPersistence.complete(projectId, message.transactionId, false); } updateUnsaved(); }, onDurable: ({ message }) => { if (projectId) { draftPersistence.complete(projectId, message.transactionId, true); } updateUnsaved(); }, onReload: (message) => { if (projectId) { draftPersistence.invalidate(projectId); } onReload?.(message); }, onUserMessage, sendApply, }); return { close() { tracker.destroy(); updateUnsaved(); ws?.close(); ws = undefined; }, connect, emit(message) { // Preserve the regular in-browser SyncEmitter contract. The builder and // canvas iframe still use this emitter for local state handshakes and // client-only stores; multiplayer only adds remote relay for server patches. events.emit("message", message); if (message.type !== "apply") { return; } if (message.transaction.object !== "server") { return; } clientSeq += 1; const applyMessage: ApplyMessage = { type: "apply", transactionId: message.transaction.id, transaction: message.transaction, clientId, actorId: clientId, clientSeq, }; if (projectId) { draftPersistence.begin(projectId, applyMessage.transactionId); } tracker.track(applyMessage); updateUnsaved(); }, on(handler) { return events.on("message", handler); }, sendPresence(payload) { const message: PresenceMessage = { type: "presence", clientId, payload, }; if (ws === undefined) { return; } ws.sendToRelay(message); }, }; };