mirror of https://github.com/requarks/wiki
NGPixel
1 month ago
parent
0c6510b958
commit
17d7b810bb
@ -0,0 +1,70 @@
|
||||
import { validate as uuidValidate } from 'uuid'
|
||||
|
||||
import { mayOnPage } from '../api/pages.ts'
|
||||
|
||||
import type { FastifyInstance, FastifyRequest } from 'fastify'
|
||||
import type { WebSocket } from 'ws'
|
||||
|
||||
/**
|
||||
* _collab Routes
|
||||
*
|
||||
* The websocket behind live collaborative editing. One socket per editor, one room per page — see
|
||||
* `core/collab.ts` for what a room is and how rooms find each other across instances.
|
||||
*
|
||||
* Unlike its neighbours under `controllers/`, nothing here is public. A room carries a page's unsaved
|
||||
* text and everyone's cursor, so joining one takes a session that may edit that page — the same
|
||||
* `write:pages` the save itself takes, checked against the page rather than against the group's
|
||||
* permission list. Whoever may only *suggest* edits does not qualify, which is what keeps a suggestion
|
||||
* the private draft it is meant to be.
|
||||
*
|
||||
* The handshake is the only place authorization happens: a socket is checked once, when it opens, and
|
||||
* a permission taken away mid-session takes effect the next time the editor is opened.
|
||||
*/
|
||||
async function routes(app: FastifyInstance) {
|
||||
app.get<{ Params: { siteId: string; pageId: string } }>(
|
||||
'/:siteId/:pageId',
|
||||
{ websocket: true, schema: { hide: true } },
|
||||
async (
|
||||
socket: WebSocket,
|
||||
req: FastifyRequest<{ Params: { siteId: string; pageId: string } }>
|
||||
) => {
|
||||
const { siteId, pageId } = req.params
|
||||
|
||||
/*
|
||||
Before the first `await`, and that is the point: the client starts talking as soon as the
|
||||
socket is open, which is well before the checks below have finished asking the database
|
||||
anything. See `capture` in `core/collab.ts`.
|
||||
*/
|
||||
const session = WIKI.collab.capture(socket)
|
||||
|
||||
/*
|
||||
Refusals close the socket rather than answering with anything: the client is y-websocket, which
|
||||
speaks the sync protocol and nothing else, and it treats a close as the signal to back off.
|
||||
The codes are in the private 4000 range, where the browser hands them to the page — which is
|
||||
how the editor tells "you may not edit this" apart from "the connection dropped" and knows not
|
||||
to reconnect. See `composables/collab.js`.
|
||||
*/
|
||||
if (!uuidValidate(siteId) || !uuidValidate(pageId)) {
|
||||
return socket.close(4400, 'Invalid site or page id')
|
||||
}
|
||||
if (!req.session?.authenticated) {
|
||||
return socket.close(4401, 'Authentication is required')
|
||||
}
|
||||
if (!WIKI.sites[siteId]?.config?.features?.collaborativeEditing) {
|
||||
return socket.close(4403, 'Collaborative editing is disabled on this site')
|
||||
}
|
||||
|
||||
const page = await WIKI.models.pages.getPage({ siteId, id: pageId })
|
||||
if (!page) {
|
||||
return socket.close(4404, 'This page does not exist')
|
||||
}
|
||||
if (!mayOnPage(req, 'write:pages', page)) {
|
||||
return socket.close(4403, 'You are not allowed to edit this page')
|
||||
}
|
||||
|
||||
await WIKI.collab.join(socket, { id: pageId, siteId }, session)
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
export default routes
|
||||
@ -0,0 +1,733 @@
|
||||
import { sql } from 'drizzle-orm'
|
||||
import * as decoding from 'lib0/decoding'
|
||||
import * as encoding from 'lib0/encoding'
|
||||
import * as awarenessProtocol from 'y-protocols/awareness'
|
||||
import * as syncProtocol from 'y-protocols/sync'
|
||||
import * as Y from 'yjs'
|
||||
|
||||
import type { PoolClient } from 'pg'
|
||||
import type { WebSocket } from 'ws'
|
||||
|
||||
/**
|
||||
* Live collaborative editing.
|
||||
*
|
||||
* A room is one page being edited by more than one person at a time. It holds a Yjs document — the
|
||||
* markdown source as a `Y.Text`, the header fields as a `Y.Map` — and the awareness state that carries
|
||||
* everyone's cursor and identity. Clients speak the y-websocket protocol to it, which is why the
|
||||
* message framing below is byte-compatible with `y-websocket`'s client rather than something of our
|
||||
* own: the browser side is that library, unmodified.
|
||||
*
|
||||
* **A room is not storage.** Nothing here is ever written back to the page — saving is still an
|
||||
* explicit act, `PATCH /pages/:id` as it always was, and a room that empties out takes any unsaved text
|
||||
* with it exactly as closing the editor always has. What a room adds is that the text survives *one*
|
||||
* participant leaving, because the others are still holding it.
|
||||
*
|
||||
* ## Across instances
|
||||
*
|
||||
* Rooms live in memory, so two people served by different instances would otherwise never meet. Their
|
||||
* updates are relayed over postgres LISTEN/NOTIFY on a channel of this module's own, separate from the
|
||||
* `wiki` channel that carries the general event bus: these are frequent, binary, and worthless a
|
||||
* second after they are sent, and none of that describes an event bus message.
|
||||
*
|
||||
* NOTIFY caps a payload at 8000 bytes, so a relayed message is base64'd and split into chunks that fit
|
||||
* — see {@link relay}. Chunks of one message arrive in order, postgres guaranteeing that much per
|
||||
* connection.
|
||||
*
|
||||
* ## Where a room's starting state comes from
|
||||
*
|
||||
* This is the one genuinely delicate part. A Yjs document cannot simply be seeded twice: two instances
|
||||
* that each insert the page's text into their own replica produce two *different* sets of operations
|
||||
* that both say "insert this text", and merging those replicas concatenates them — the document ends
|
||||
* up holding the page twice. So a room being created asks the cluster first ({@link peerState}) and
|
||||
* only falls back to the stored page when nobody answers.
|
||||
*
|
||||
* Two instances cold-starting the same room in the same instant would still both fall back, so that
|
||||
* seed is made *deterministic*: it is built in a scratch document pinned to client id 0, and two seeds
|
||||
* of identical text therefore produce byte-identical operations, which merge as one. That is also what
|
||||
* lets a client reconnect after a network blip and push back the edits it made while it was away — its
|
||||
* local copy of the seed is the same seed a freshly created room builds.
|
||||
*/
|
||||
|
||||
/** y-websocket message types. The values are that protocol's, not ours. */
|
||||
const MESSAGE_SYNC = 0
|
||||
const MESSAGE_AWARENESS = 1
|
||||
|
||||
const NOTIFY_CHANNEL = 'wiki_collab'
|
||||
|
||||
/**
|
||||
* Base64 characters per NOTIFY payload. Postgres refuses a payload over 8000 bytes, and the JSON
|
||||
* envelope around the chunk fits comfortably in the slack this leaves.
|
||||
*/
|
||||
const RELAY_CHUNK_SIZE = 5000
|
||||
|
||||
/** How long a half-assembled relay message waits for the rest of its chunks before being dropped. */
|
||||
const RELAY_REASSEMBLY_TIMEOUT = 10 * 1000
|
||||
|
||||
/**
|
||||
* How long a new room waits for a peer to hand over the state it already has, before seeding itself
|
||||
* from the stored page. Only paid when this instance does not already have the room open, and skipped
|
||||
* entirely when no other instance is running — which is the ordinary case.
|
||||
*/
|
||||
const PEER_STATE_TIMEOUT = 500
|
||||
|
||||
/** How long the "is anyone else running?" answer is trusted before it is looked up again. */
|
||||
const PEER_PRESENCE_TTL = 15 * 1000
|
||||
|
||||
/** Keepalive interval. An idle websocket is what a reverse proxy cuts first. */
|
||||
const PING_INTERVAL = 30 * 1000
|
||||
|
||||
/**
|
||||
* Marks a document or awareness change as having arrived over the relay, so that applying it here does
|
||||
* not send it straight back out to the instance it came from.
|
||||
*/
|
||||
const RELAYED = Symbol('collabRelayed')
|
||||
|
||||
interface CollabConn {
|
||||
/** Awareness client ids this socket is responsible for, so a disconnect can retract exactly those. */
|
||||
clients: Set<number>
|
||||
/** Answered the last keepalive ping. */
|
||||
alive: boolean
|
||||
}
|
||||
|
||||
interface CollabSession {
|
||||
/** The room this socket ended up in, or null while it is still being decided. */
|
||||
room: CollabRoom | null
|
||||
/** Frames that arrived before there was a room to hand them to. */
|
||||
pending: Uint8Array[]
|
||||
}
|
||||
|
||||
interface CollabRoom {
|
||||
pageId: string
|
||||
siteId: string
|
||||
doc: Y.Doc
|
||||
awareness: awarenessProtocol.Awareness
|
||||
conns: Map<WebSocket, CollabConn>
|
||||
/** Resolves once the document holds its starting state and clients may be synced against it. */
|
||||
ready: Promise<void>
|
||||
/** Whether this room is still filling itself, i.e. has nothing worth handing to a peer yet. */
|
||||
provisional: boolean
|
||||
}
|
||||
|
||||
interface SaveInfo {
|
||||
versionDate: string
|
||||
authorId: string
|
||||
authorName: string
|
||||
}
|
||||
|
||||
interface RelayEnvelope {
|
||||
/** Instance the message came from. */
|
||||
i: string
|
||||
/** Room, i.e. page id. */
|
||||
r: string
|
||||
t: 'update' | 'awareness' | 'hello' | 'state' | 'saved'
|
||||
/** Payload: base64 for the binary kinds, JSON for `saved`, absent for `hello`. */
|
||||
p?: string
|
||||
/** Instance this is addressed to, when it is a reply rather than a broadcast. */
|
||||
to?: string
|
||||
/** Chunking: message id, chunk index, chunk count. Absent on a message that fits in one. */
|
||||
m?: string
|
||||
c?: number
|
||||
n?: number
|
||||
}
|
||||
|
||||
interface PartialRelay {
|
||||
parts: (string | undefined)[]
|
||||
remaining: number
|
||||
timer: NodeJS.Timeout
|
||||
}
|
||||
|
||||
/**
|
||||
* A websocket frame as bytes, whatever shape `ws` handed it over in.
|
||||
*
|
||||
* A fragmented message arrives as an array of buffers, and a whole one as a single `Buffer` — which is
|
||||
* a view into a larger pool, so its offset and length matter. The result is a view over that same
|
||||
* memory and is only safe to read during the event that delivered it; anything held on to has to be
|
||||
* copied first.
|
||||
*/
|
||||
function toBytes(data: unknown): Uint8Array {
|
||||
if (Array.isArray(data)) {
|
||||
return new Uint8Array(Buffer.concat(data))
|
||||
}
|
||||
if (Buffer.isBuffer(data)) {
|
||||
return new Uint8Array(data.buffer, data.byteOffset, data.byteLength)
|
||||
}
|
||||
return new Uint8Array(data as ArrayBuffer)
|
||||
}
|
||||
|
||||
/**
|
||||
* The state a room starts from when it has to build one itself, as a Yjs update.
|
||||
*
|
||||
* Built in a scratch document whose client id is pinned to 0, so that the bytes depend on nothing but
|
||||
* the page — see the note at the top of this file on why that matters.
|
||||
*/
|
||||
function buildSeed(page: {
|
||||
content?: string | null
|
||||
title?: string | null
|
||||
description?: string | null
|
||||
icon?: string | null
|
||||
}): Uint8Array {
|
||||
const seed = new Y.Doc()
|
||||
seed.clientID = 0
|
||||
seed.transact(() => {
|
||||
seed.getText('content').insert(0, page.content ?? '')
|
||||
const props = seed.getMap('props')
|
||||
props.set('title', page.title ?? '')
|
||||
props.set('description', page.description ?? '')
|
||||
props.set('icon', page.icon ?? '')
|
||||
})
|
||||
const update = Y.encodeStateAsUpdate(seed)
|
||||
seed.destroy()
|
||||
return update
|
||||
}
|
||||
|
||||
export default {
|
||||
rooms: new Map<string, CollabRoom>(),
|
||||
listenClient: null as PoolClient | null,
|
||||
/** Chunked relay messages still waiting for the rest of themselves, keyed by sender and message id. */
|
||||
partials: new Map<string, PartialRelay>(),
|
||||
/** Rooms this instance is waiting on a peer's state for, by page id. */
|
||||
awaitingState: new Map<string, (update: Uint8Array) => void>(),
|
||||
relaySeq: 0,
|
||||
peerPresence: { known: false, checkedAt: 0 },
|
||||
pingTimer: null as NodeJS.Timeout | null,
|
||||
|
||||
/**
|
||||
* Open the relay connection.
|
||||
*
|
||||
* A client of its own rather than the event bus's: these messages are far more frequent than events
|
||||
* are, and a slow consumer on one channel should not hold up the other.
|
||||
*/
|
||||
async init(): Promise<void> {
|
||||
this.listenClient = await WIKI.dbManager.pool!.connect()
|
||||
await this.listenClient.query(`SET application_name = 'Wiki.js - ${WIKI.INSTANCE_ID}:COLLAB'`)
|
||||
this.listenClient.on('notification', (msg) => {
|
||||
if (msg.channel !== NOTIFY_CHANNEL || !msg.payload) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
this.receiveRelay(JSON.parse(msg.payload) as RelayEnvelope)
|
||||
} catch (err: any) {
|
||||
WIKI.logger.warn(`Malformed collaboration relay message: ${err.message}`)
|
||||
}
|
||||
})
|
||||
await this.listenClient.query(`LISTEN ${NOTIFY_CHANNEL}`)
|
||||
|
||||
this.pingTimer = setInterval(() => {
|
||||
for (const room of this.rooms.values()) {
|
||||
for (const [conn, state] of room.conns) {
|
||||
// -> A socket whose peer stopped answering is dropped by the `close` handler that
|
||||
// `terminate()` triggers, which is also what takes its cursor off everyone's screen
|
||||
if (!state.alive) {
|
||||
conn.terminate()
|
||||
continue
|
||||
}
|
||||
state.alive = false
|
||||
try {
|
||||
conn.ping()
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
}, PING_INTERVAL)
|
||||
|
||||
WIKI.logger.info('Collaborative editing initialized successfully: [ OK ]')
|
||||
},
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
if (this.pingTimer) {
|
||||
clearInterval(this.pingTimer)
|
||||
this.pingTimer = null
|
||||
}
|
||||
for (const partial of this.partials.values()) {
|
||||
clearTimeout(partial.timer)
|
||||
}
|
||||
this.partials.clear()
|
||||
for (const room of this.rooms.values()) {
|
||||
for (const conn of room.conns.keys()) {
|
||||
conn.close(1001, 'Server is shutting down')
|
||||
}
|
||||
room.awareness.destroy()
|
||||
room.doc.destroy()
|
||||
}
|
||||
this.rooms.clear()
|
||||
if (this.listenClient) {
|
||||
this.listenClient.release(true)
|
||||
this.listenClient = null
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Whether another instance is currently running.
|
||||
*
|
||||
* Asked so that the single-instance case — very much the common one — does not spend
|
||||
* {@link PEER_STATE_TIMEOUT} waiting for an answer that cannot come. Instances are not registered
|
||||
* anywhere, so this reads what the admin area's instance list reads: our own connections name
|
||||
* themselves in `pg_stat_activity`.
|
||||
*/
|
||||
async hasPeers(): Promise<boolean> {
|
||||
const now = Date.now()
|
||||
if (now - this.peerPresence.checkedAt < PEER_PRESENCE_TTL) {
|
||||
return this.peerPresence.known
|
||||
}
|
||||
const ownName = `Wiki.js - ${WIKI.INSTANCE_ID}:COLLAB`
|
||||
try {
|
||||
const result = await WIKI.db.execute(
|
||||
sql`SELECT 1 FROM pg_stat_activity WHERE datname = current_database()
|
||||
AND application_name LIKE 'Wiki.js - %:COLLAB'
|
||||
AND application_name <> ${ownName} LIMIT 1`
|
||||
)
|
||||
this.peerPresence = { known: result.rows.length > 0, checkedAt: now }
|
||||
} catch (err: any) {
|
||||
// -> Assume company: waiting 500ms is a far smaller mistake than duplicating a page's text
|
||||
WIKI.logger.warn(`Could not determine whether other instances are running: ${err.message}`)
|
||||
this.peerPresence = { known: true, checkedAt: now }
|
||||
}
|
||||
return this.peerPresence.known
|
||||
},
|
||||
|
||||
/**
|
||||
* Start listening to a socket before anything is known about it.
|
||||
*
|
||||
* Called the instant the socket opens, and synchronously — the client does not wait to be welcomed.
|
||||
* y-websocket sends its first sync message immediately, while the route is still away asking the
|
||||
* database whether this user may edit this page at all, and an event nobody is listening for is
|
||||
* simply gone. That one message is the whole handshake: miss it and the client sits there holding an
|
||||
* empty document, because it is never going to ask twice.
|
||||
*
|
||||
* So the frames are collected here and replayed by {@link join} once there is a room to put them to.
|
||||
*/
|
||||
capture(conn: WebSocket): CollabSession {
|
||||
const session: CollabSession = { room: null, pending: [] }
|
||||
conn.on('message', (data: unknown) => {
|
||||
if (session.room) {
|
||||
this.onMessage(session.room, conn, toBytes(data))
|
||||
} else {
|
||||
// -> Copied, not referenced: `toBytes` hands back a view into a buffer `ws` owns, which is
|
||||
// only good for the length of this event
|
||||
session.pending.push(new Uint8Array(toBytes(data)))
|
||||
}
|
||||
})
|
||||
conn.on('close', () => {
|
||||
if (session.room) {
|
||||
this.onClose(session.room, conn)
|
||||
}
|
||||
})
|
||||
conn.on('error', (err: Error) => {
|
||||
WIKI.logger.debug(`Collaboration socket error: ${err.message}`)
|
||||
})
|
||||
return session
|
||||
},
|
||||
|
||||
/**
|
||||
* Put a socket into a page's room, syncing it against whatever state that room holds.
|
||||
*
|
||||
* The caller is responsible for having decided that this user may edit this page — see
|
||||
* `controllers/collab.ts`. Nothing below re-checks it.
|
||||
*/
|
||||
async join(
|
||||
conn: WebSocket,
|
||||
page: { id: string; siteId: string },
|
||||
session: CollabSession
|
||||
): Promise<void> {
|
||||
/*
|
||||
Asked for repeatedly, because a room can be dropped while this socket was waiting for it: another
|
||||
socket that gave up during the same setup takes the still-empty room down with it. Joining that
|
||||
one would put this editor in a room nothing else can find.
|
||||
*/
|
||||
let room = await this.ensureRoom(page)
|
||||
for (let attempt = 0; this.rooms.get(page.id) !== room && attempt < 3; attempt++) {
|
||||
room = await this.ensureRoom(page)
|
||||
}
|
||||
|
||||
// -> The socket may well have gone away while the room was being set up
|
||||
if (conn.readyState !== conn.OPEN) {
|
||||
this.closeRoomIfEmpty(room)
|
||||
return
|
||||
}
|
||||
|
||||
const state: CollabConn = { clients: new Set(), alive: true }
|
||||
room.conns.set(conn, state)
|
||||
conn.on('pong', () => {
|
||||
state.alive = true
|
||||
})
|
||||
session.room = room
|
||||
|
||||
// -> Sync step 1: what this room has, so the client can say what it is missing
|
||||
const syncEncoder = encoding.createEncoder()
|
||||
encoding.writeVarUint(syncEncoder, MESSAGE_SYNC)
|
||||
syncProtocol.writeSyncStep1(syncEncoder, room.doc)
|
||||
this.send(conn, encoding.toUint8Array(syncEncoder))
|
||||
|
||||
// -> And everyone already in the room, so their cursors are there from the first frame
|
||||
const states = room.awareness.getStates()
|
||||
if (states.size > 0) {
|
||||
const awarenessEncoder = encoding.createEncoder()
|
||||
encoding.writeVarUint(awarenessEncoder, MESSAGE_AWARENESS)
|
||||
encoding.writeVarUint8Array(
|
||||
awarenessEncoder,
|
||||
awarenessProtocol.encodeAwarenessUpdate(room.awareness, [...states.keys()])
|
||||
)
|
||||
this.send(conn, encoding.toUint8Array(awarenessEncoder))
|
||||
}
|
||||
|
||||
for (const message of session.pending) {
|
||||
this.onMessage(room, conn, message)
|
||||
}
|
||||
session.pending = []
|
||||
},
|
||||
|
||||
/**
|
||||
* The room for a page, creating and populating it if this instance does not have it open.
|
||||
*
|
||||
* Concurrent joiners share one room *and one initialization*: the room goes into the map before it
|
||||
* has any state, and `ready` is what everything else waits on.
|
||||
*/
|
||||
async ensureRoom(page: { id: string; siteId: string }): Promise<CollabRoom> {
|
||||
const existing = this.rooms.get(page.id)
|
||||
if (existing) {
|
||||
await existing.ready
|
||||
return existing
|
||||
}
|
||||
|
||||
const doc = new Y.Doc()
|
||||
const awareness = new awarenessProtocol.Awareness(doc)
|
||||
// -> The server is not a participant. Left as it comes, its own empty state would show up in the
|
||||
// room as a cursor nobody owns, and be relayed to every other instance as one.
|
||||
awareness.setLocalState(null)
|
||||
|
||||
const room: CollabRoom = {
|
||||
pageId: page.id,
|
||||
siteId: page.siteId,
|
||||
doc,
|
||||
awareness,
|
||||
conns: new Map(),
|
||||
ready: Promise.resolve(),
|
||||
provisional: true
|
||||
}
|
||||
this.rooms.set(page.id, room)
|
||||
|
||||
doc.on('update', (update: Uint8Array, origin: unknown) => {
|
||||
const encoder = encoding.createEncoder()
|
||||
encoding.writeVarUint(encoder, MESSAGE_SYNC)
|
||||
syncProtocol.writeUpdate(encoder, update)
|
||||
const message = encoding.toUint8Array(encoder)
|
||||
for (const conn of room.conns.keys()) {
|
||||
this.send(conn, message)
|
||||
}
|
||||
if (origin !== RELAYED) {
|
||||
this.relay({ r: room.pageId, t: 'update', p: Buffer.from(update).toString('base64') })
|
||||
}
|
||||
})
|
||||
|
||||
awareness.on(
|
||||
'update',
|
||||
(
|
||||
{ added, updated, removed }: { added: number[]; updated: number[]; removed: number[] },
|
||||
origin: unknown
|
||||
) => {
|
||||
const changed = [...added, ...updated, ...removed]
|
||||
// -> Remember whose cursors these are, so that a disconnect can retract exactly them
|
||||
const owner = room.conns.get(origin as WebSocket)
|
||||
if (owner) {
|
||||
for (const clientId of added) {
|
||||
owner.clients.add(clientId)
|
||||
}
|
||||
for (const clientId of removed) {
|
||||
owner.clients.delete(clientId)
|
||||
}
|
||||
}
|
||||
const update = awarenessProtocol.encodeAwarenessUpdate(awareness, changed)
|
||||
const encoder = encoding.createEncoder()
|
||||
encoding.writeVarUint(encoder, MESSAGE_AWARENESS)
|
||||
encoding.writeVarUint8Array(encoder, update)
|
||||
const message = encoding.toUint8Array(encoder)
|
||||
for (const conn of room.conns.keys()) {
|
||||
this.send(conn, message)
|
||||
}
|
||||
if (origin !== RELAYED) {
|
||||
this.relay({
|
||||
r: room.pageId,
|
||||
t: 'awareness',
|
||||
p: Buffer.from(update).toString('base64')
|
||||
})
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
room.ready = this.initRoom(room)
|
||||
await room.ready
|
||||
return room
|
||||
},
|
||||
|
||||
/**
|
||||
* Fill a newly created room with the state it should start from: a peer's copy if the cluster
|
||||
* already has this page open, and the stored page if not.
|
||||
*/
|
||||
async initRoom(room: CollabRoom): Promise<void> {
|
||||
try {
|
||||
const fromPeer = (await this.hasPeers()) ? await this.peerState(room.pageId) : null
|
||||
if (fromPeer) {
|
||||
Y.applyUpdate(room.doc, fromPeer, RELAYED)
|
||||
} else {
|
||||
const page = await WIKI.models.pages.getPage({
|
||||
siteId: room.siteId,
|
||||
id: room.pageId,
|
||||
withContent: true
|
||||
})
|
||||
// -> A page that went away between the permission check and here leaves an empty room, which
|
||||
// the first disconnect clears away again
|
||||
Y.applyUpdate(room.doc, buildSeed(page ?? {}), RELAYED)
|
||||
}
|
||||
} catch (err: any) {
|
||||
WIKI.logger.warn(
|
||||
`Failed to initialize the collaboration room for page ${room.pageId}: ${err.message}`
|
||||
)
|
||||
} finally {
|
||||
room.provisional = false
|
||||
this.awaitingState.delete(room.pageId)
|
||||
}
|
||||
},
|
||||
|
||||
/** Ask the cluster for a room's current state, resolving to null if nobody answers in time. */
|
||||
peerState(pageId: string): Promise<Uint8Array | null> {
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.awaitingState.delete(pageId)
|
||||
resolve(null)
|
||||
}, PEER_STATE_TIMEOUT)
|
||||
this.awaitingState.set(pageId, (update) => {
|
||||
clearTimeout(timer)
|
||||
this.awaitingState.delete(pageId)
|
||||
resolve(update)
|
||||
})
|
||||
this.relay({ r: pageId, t: 'hello' })
|
||||
})
|
||||
},
|
||||
|
||||
onMessage(room: CollabRoom, conn: WebSocket, message: Uint8Array): void {
|
||||
try {
|
||||
const decoder = decoding.createDecoder(message)
|
||||
const encoder = encoding.createEncoder()
|
||||
switch (decoding.readVarUint(decoder)) {
|
||||
case MESSAGE_SYNC: {
|
||||
encoding.writeVarUint(encoder, MESSAGE_SYNC)
|
||||
// -> The socket is the origin, which is how the awareness bookkeeping above knows whose
|
||||
// cursors an update carries
|
||||
syncProtocol.readSyncMessage(decoder, encoder, room.doc, conn)
|
||||
if (encoding.length(encoder) > 1) {
|
||||
this.send(conn, encoding.toUint8Array(encoder))
|
||||
}
|
||||
break
|
||||
}
|
||||
case MESSAGE_AWARENESS: {
|
||||
awarenessProtocol.applyAwarenessUpdate(
|
||||
room.awareness,
|
||||
decoding.readVarUint8Array(decoder),
|
||||
conn
|
||||
)
|
||||
break
|
||||
}
|
||||
}
|
||||
} catch (err: any) {
|
||||
WIKI.logger.warn(
|
||||
`Failed to handle a collaboration message on page ${room.pageId}: ${err.message}`
|
||||
)
|
||||
}
|
||||
},
|
||||
|
||||
onClose(room: CollabRoom, conn: WebSocket): void {
|
||||
const state = room.conns.get(conn)
|
||||
room.conns.delete(conn)
|
||||
if (state && state.clients.size > 0) {
|
||||
// -> Announced as an awareness change, which is what takes the avatar out of the header and the
|
||||
// cursor out of the text for everyone else, here and on every other instance
|
||||
awarenessProtocol.removeAwarenessStates(room.awareness, [...state.clients], null)
|
||||
}
|
||||
this.closeRoomIfEmpty(room)
|
||||
},
|
||||
|
||||
/**
|
||||
* Drop a room nobody on this instance is in.
|
||||
*
|
||||
* Immediately, with no grace period: an editor closed without saving has always lost its unsaved
|
||||
* text, and a room outliving its last participant would quietly resurrect it on the next visit.
|
||||
* Discarding an edit is that same act and needs nothing of its own — the socket closes and the state
|
||||
* goes with it.
|
||||
*
|
||||
* Peers are not told. A room elsewhere is a replica in its own right whose participants are still
|
||||
* editing; this instance simply asks for their state again next time someone here opens the page.
|
||||
*/
|
||||
closeRoomIfEmpty(room: CollabRoom): void {
|
||||
if (room.conns.size > 0 || this.rooms.get(room.pageId) !== room) {
|
||||
return
|
||||
}
|
||||
this.rooms.delete(room.pageId)
|
||||
room.awareness.destroy()
|
||||
room.doc.destroy()
|
||||
},
|
||||
|
||||
/**
|
||||
* Tell everyone editing a page that it has just been saved.
|
||||
*
|
||||
* Written into the document rather than sent as a message of its own, so that it reaches the other
|
||||
* instances the way an edit does and a client joining a moment later sees the same thing. Nothing
|
||||
* about the text changes — this only tells the other editors that what they are looking at is now
|
||||
* what is stored, and their Save button can go quiet.
|
||||
*
|
||||
* The save does not necessarily land on an instance that has the room, so an instance without one
|
||||
* passes the news along instead.
|
||||
*/
|
||||
pageSaved(pageId: string, info: SaveInfo): void {
|
||||
const room = this.rooms.get(pageId)
|
||||
if (room) {
|
||||
room.doc.getMap('meta').set('lastSave', info)
|
||||
} else {
|
||||
this.relay({ r: pageId, t: 'saved', p: JSON.stringify(info) })
|
||||
}
|
||||
},
|
||||
|
||||
// ----------------------------------------
|
||||
// Relay
|
||||
// ----------------------------------------
|
||||
|
||||
/** Publish a message to the other instances, split into chunks postgres will accept. */
|
||||
relay(message: Omit<RelayEnvelope, 'i'>): void {
|
||||
if (!this.listenClient) {
|
||||
return
|
||||
}
|
||||
const envelope: RelayEnvelope = { ...message, i: WIKI.INSTANCE_ID }
|
||||
const payload = envelope.p
|
||||
if (!payload || payload.length <= RELAY_CHUNK_SIZE) {
|
||||
this.publish(envelope)
|
||||
return
|
||||
}
|
||||
const count = Math.ceil(payload.length / RELAY_CHUNK_SIZE)
|
||||
const messageId = `${this.relaySeq++}`
|
||||
for (let index = 0; index < count; index++) {
|
||||
this.publish({
|
||||
...envelope,
|
||||
p: payload.slice(index * RELAY_CHUNK_SIZE, (index + 1) * RELAY_CHUNK_SIZE),
|
||||
m: messageId,
|
||||
c: index,
|
||||
n: count
|
||||
})
|
||||
}
|
||||
},
|
||||
|
||||
publish(envelope: RelayEnvelope): void {
|
||||
this.listenClient
|
||||
?.query('SELECT pg_notify($1, $2)', [NOTIFY_CHANNEL, JSON.stringify(envelope)])
|
||||
.catch((err: any) => {
|
||||
WIKI.logger.warn(`Failed to relay a collaboration message: ${err.message}`)
|
||||
})
|
||||
},
|
||||
|
||||
receiveRelay(envelope: RelayEnvelope): void {
|
||||
if (envelope.i === WIKI.INSTANCE_ID) {
|
||||
return
|
||||
}
|
||||
if (envelope.to && envelope.to !== WIKI.INSTANCE_ID) {
|
||||
return
|
||||
}
|
||||
if (envelope.m !== undefined && envelope.n !== undefined) {
|
||||
const assembled = this.reassemble(envelope)
|
||||
if (assembled === null) {
|
||||
return
|
||||
}
|
||||
envelope.p = assembled
|
||||
}
|
||||
switch (envelope.t) {
|
||||
case 'hello': {
|
||||
// -> Somewhere else is opening this page and has nothing yet. Only a room that is past its own
|
||||
// setup is worth answering with; one still filling itself would hand over an empty document.
|
||||
const room = this.rooms.get(envelope.r)
|
||||
if (!room || room.provisional) {
|
||||
return
|
||||
}
|
||||
this.relay({
|
||||
r: envelope.r,
|
||||
t: 'state',
|
||||
to: envelope.i,
|
||||
p: Buffer.from(Y.encodeStateAsUpdate(room.doc)).toString('base64')
|
||||
})
|
||||
break
|
||||
}
|
||||
case 'state': {
|
||||
const waiting = this.awaitingState.get(envelope.r)
|
||||
if (!waiting) {
|
||||
// -> Too late to be adopted, and merging it now is exactly the duplication this handshake
|
||||
// exists to avoid. See the note at the top of this file.
|
||||
WIKI.logger.debug(
|
||||
`Ignoring a late collaboration state for page ${envelope.r} from instance ${envelope.i}`
|
||||
)
|
||||
return
|
||||
}
|
||||
waiting(Buffer.from(envelope.p ?? '', 'base64'))
|
||||
break
|
||||
}
|
||||
case 'update': {
|
||||
const room = this.rooms.get(envelope.r)
|
||||
if (room) {
|
||||
Y.applyUpdate(room.doc, Buffer.from(envelope.p ?? '', 'base64'), RELAYED)
|
||||
}
|
||||
break
|
||||
}
|
||||
case 'awareness': {
|
||||
const room = this.rooms.get(envelope.r)
|
||||
if (room) {
|
||||
awarenessProtocol.applyAwarenessUpdate(
|
||||
room.awareness,
|
||||
Buffer.from(envelope.p ?? '', 'base64'),
|
||||
RELAYED
|
||||
)
|
||||
}
|
||||
break
|
||||
}
|
||||
case 'saved': {
|
||||
const room = this.rooms.get(envelope.r)
|
||||
if (room && envelope.p) {
|
||||
room.doc.getMap('meta').set('lastSave', JSON.parse(envelope.p) as SaveInfo)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
/** Collect a chunked message, returning the whole payload once the last chunk lands. */
|
||||
reassemble(envelope: RelayEnvelope): string | null {
|
||||
const key = `${envelope.i}:${envelope.m}`
|
||||
let partial = this.partials.get(key)
|
||||
if (!partial) {
|
||||
partial = {
|
||||
parts: Array.from({ length: envelope.n! }),
|
||||
remaining: envelope.n!,
|
||||
timer: setTimeout(() => {
|
||||
// -> An instance that died mid-message would otherwise leave its chunks here for good
|
||||
this.partials.delete(key)
|
||||
}, RELAY_REASSEMBLY_TIMEOUT)
|
||||
}
|
||||
this.partials.set(key, partial)
|
||||
}
|
||||
if (partial.parts[envelope.c!] !== undefined) {
|
||||
return null
|
||||
}
|
||||
partial.parts[envelope.c!] = envelope.p ?? ''
|
||||
partial.remaining--
|
||||
if (partial.remaining > 0) {
|
||||
return null
|
||||
}
|
||||
clearTimeout(partial.timer)
|
||||
this.partials.delete(key)
|
||||
return partial.parts.join('')
|
||||
},
|
||||
|
||||
send(conn: WebSocket, message: Uint8Array): void {
|
||||
if (conn.readyState !== conn.OPEN) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
conn.send(message)
|
||||
} catch {
|
||||
conn.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,216 @@
|
||||
<template>
|
||||
<!--
|
||||
Nothing at all when you are on your own, which is the ordinary case: a single bubble of your own
|
||||
face says nothing you did not already know, and the header has better uses for the space.
|
||||
-->
|
||||
<div
|
||||
v-if="collabStore.people.length > 1"
|
||||
class="collab-presence"
|
||||
role="group"
|
||||
:aria-label="t('editor.collab.participants')">
|
||||
<!--
|
||||
The bubble is wrapped rather than styled alone because it has to clip the avatar to a circle,
|
||||
and a ring rippling outwards from something that clips its own children would be cut off at the
|
||||
edge it is supposed to leave.
|
||||
-->
|
||||
<div
|
||||
v-for="person of visible"
|
||||
:key="person.id"
|
||||
class="collab-presence-person"
|
||||
:class="{ 'is-typing': person.typing }">
|
||||
<span
|
||||
class="collab-presence-wave"
|
||||
:style="{ borderColor: person.color }"
|
||||
aria-hidden="true" />
|
||||
<div class="collab-presence-bubble" :style="{ backgroundColor: person.color }">
|
||||
<!--
|
||||
No `alt`: the name is already on the group's label and in the tooltip, and an avatar that
|
||||
fails to load should fall back to the coloured circle rather than to the person's name in
|
||||
plain text across the header.
|
||||
-->
|
||||
<img v-if="person.hasAvatar" :src="`/_user/${person.id}/avatar`" alt="" />
|
||||
<span v-else>{{ initials(person.name) }}</span>
|
||||
</div>
|
||||
<w-tooltip>
|
||||
{{ personLabel(person) }}
|
||||
</w-tooltip>
|
||||
</div>
|
||||
<!--
|
||||
The count pulses on behalf of whoever it is standing in for, so that someone typing out of sight
|
||||
is not simply invisible.
|
||||
-->
|
||||
<div
|
||||
v-if="overflow > 0"
|
||||
class="collab-presence-person collab-presence-person--overflow"
|
||||
:class="{ 'is-typing': overflowTyping }">
|
||||
<span class="collab-presence-wave" aria-hidden="true" />
|
||||
<div class="collab-presence-bubble collab-presence-overflow">+{{ overflow }}</div>
|
||||
<w-tooltip>{{ overflowNames }}</w-tooltip>
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
<script setup>
|
||||
import { computed } from 'vue'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
|
||||
import { useCollabStore } from '@/stores/collab'
|
||||
|
||||
/**
|
||||
* Who is editing this page right now, as a row of overlapping faces in the page header.
|
||||
*
|
||||
* Fed entirely by `stores/collab.js`, so it is empty whenever there is no session — which covers the
|
||||
* site having the feature off, the editor being anything other than markdown, and an edit being
|
||||
* suggested rather than made.
|
||||
*/
|
||||
|
||||
const collabStore = useCollabStore()
|
||||
|
||||
const { t } = useI18n()
|
||||
|
||||
/** Past this many the row starts costing more space than it is worth, and the rest become a count. */
|
||||
const MAX_VISIBLE = 4
|
||||
|
||||
const visible = computed(() => collabStore.people.slice(0, MAX_VISIBLE))
|
||||
const hidden = computed(() => collabStore.people.slice(MAX_VISIBLE))
|
||||
const overflow = computed(() => hidden.value.length)
|
||||
const overflowTyping = computed(() => hidden.value.some((person) => person.typing))
|
||||
const overflowNames = computed(() => hidden.value.map(personLabel).join(', '))
|
||||
|
||||
/** A participant's name, or `You` for the reader's own face. */
|
||||
function personLabel(person) {
|
||||
return person.isSelf ? t('editor.collab.you') : person.name
|
||||
}
|
||||
|
||||
/**
|
||||
* Up to two letters, from the first and last word of the name — `Ada Lovelace` gives `AL`, and a
|
||||
* mononym gives its first letter. Falls back to a neutral glyph rather than an empty circle for an
|
||||
* account with no name on it.
|
||||
*/
|
||||
function initials(name) {
|
||||
const words = (name ?? '').trim().split(/\s+/).filter(Boolean)
|
||||
if (words.length < 1) {
|
||||
return '?'
|
||||
}
|
||||
const first = words[0][0]
|
||||
const last = words.length > 1 ? words.at(-1)[0] : ''
|
||||
return `${first}${last}`.toUpperCase()
|
||||
}
|
||||
</script>
|
||||
|
||||
<style scoped lang="scss">
|
||||
.collab-presence {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
/* -> Leaves the leftmost bubble's own overlap margin with nothing to overlap into */
|
||||
padding-left: 8px;
|
||||
|
||||
&-person {
|
||||
position: relative;
|
||||
/* -> The overlap that makes the row read as a group rather than a list of separate faces */
|
||||
margin-left: -8px;
|
||||
}
|
||||
|
||||
/*
|
||||
The ripple. Sits under the faces rather than over them, so a wave passing beneath the next avatar
|
||||
along does not wash over it -- `z-index: 0` against the bubbles' `1`, in document order, is what
|
||||
puts it there.
|
||||
*/
|
||||
&-wave {
|
||||
position: absolute;
|
||||
z-index: 0;
|
||||
inset: 0;
|
||||
border-radius: 9999px;
|
||||
/* -> A hairline: the ring is meant to be noticed out of the corner of an eye, not read */
|
||||
border: 1px solid transparent;
|
||||
opacity: 0;
|
||||
pointer-events: none;
|
||||
}
|
||||
|
||||
&-person.is-typing &-wave {
|
||||
animation: collab-presence-wave 1.6s ease-out infinite;
|
||||
}
|
||||
|
||||
&-bubble {
|
||||
position: relative;
|
||||
z-index: 1;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
justify-content: center;
|
||||
width: 30px;
|
||||
height: 30px;
|
||||
border-radius: 9999px;
|
||||
color: #fff;
|
||||
font-size: 0.7rem;
|
||||
font-weight: 600;
|
||||
letter-spacing: 0.02em;
|
||||
overflow: hidden;
|
||||
user-select: none;
|
||||
|
||||
/*
|
||||
The ring is what stops two adjacent faces from reading as one shape, so it has to be the header
|
||||
behind them rather than a fixed colour — the header is near-white on one theme and near-black
|
||||
on the other.
|
||||
*/
|
||||
@at-root .body--light & {
|
||||
box-shadow: 0 0 0 2px $grey-1;
|
||||
}
|
||||
@at-root .body--dark & {
|
||||
box-shadow: 0 0 0 2px $dark-3;
|
||||
}
|
||||
|
||||
img {
|
||||
width: 100%;
|
||||
height: 100%;
|
||||
object-fit: cover;
|
||||
}
|
||||
}
|
||||
|
||||
&-overflow {
|
||||
@at-root .body--light & {
|
||||
background-color: $grey-6;
|
||||
}
|
||||
@at-root .body--dark & {
|
||||
background-color: $grey-8;
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
The count stands in for several people at once and so has no one colour to ripple in; it borrows
|
||||
the grey it is drawn in. Every other wave takes its colour from its owner, inline.
|
||||
*/
|
||||
&-person--overflow &-wave {
|
||||
@at-root .body--light & {
|
||||
border-color: $grey-6;
|
||||
}
|
||||
@at-root .body--dark & {
|
||||
border-color: $grey-8;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@keyframes collab-presence-wave {
|
||||
0% {
|
||||
transform: scale(1);
|
||||
opacity: 0.35;
|
||||
}
|
||||
100% {
|
||||
transform: scale(2);
|
||||
opacity: 0;
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
A ring that never stops moving is exactly what someone who asked for less motion asked to be spared,
|
||||
and the information it carries -- who is typing -- would be lost with it. So it stops expanding and
|
||||
stays put instead: a steady halo that still says the same thing.
|
||||
*/
|
||||
@media (prefers-reduced-motion: reduce) {
|
||||
.collab-presence-person.is-typing .collab-presence-wave {
|
||||
animation: none;
|
||||
transform: scale(1.35);
|
||||
/* -> Held a little stronger than the moving ring, having only stillness to be noticed by */
|
||||
opacity: 0.45;
|
||||
}
|
||||
}
|
||||
</style>
|
||||
@ -0,0 +1,454 @@
|
||||
import { watch } from 'vue'
|
||||
|
||||
import { MonacoBinding } from 'y-monaco'
|
||||
import { WebsocketProvider } from 'y-websocket'
|
||||
import * as Y from 'yjs'
|
||||
|
||||
import { useCollabStore } from '@/stores/collab'
|
||||
import { useEditorStore } from '@/stores/editor'
|
||||
import { usePageStore } from '@/stores/page'
|
||||
import { useUserStore } from '@/stores/user'
|
||||
|
||||
/**
|
||||
* Live collaborative editing, browser side.
|
||||
*
|
||||
* One session at a time — there is one editor open at a time — so this is a module singleton rather
|
||||
* than a per-component composable. The Yjs document, the websocket and the Monaco binding are held
|
||||
* here, deliberately outside of Vue's reactivity: a CRDT is a graph of mutable nodes and wrapping one
|
||||
* in a proxy is both pointless and slow. What components need is mirrored into `stores/collab.js`.
|
||||
*
|
||||
* What is shared is the markdown source and the three fields in the page header. Everything else about
|
||||
* a page — its tags, its path, the properties panel — is not, and the last save wins on those, exactly
|
||||
* as it did before any of this existed.
|
||||
*
|
||||
* Saving is unchanged and still explicit. All this session does about it is listen: the server writes
|
||||
* the fact of a save into the document, and the editors that did not make it stop calling themselves
|
||||
* unsaved.
|
||||
*/
|
||||
|
||||
/** How long to wait for the first sync before giving up and letting the author type offline. */
|
||||
const SYNC_TIMEOUT = 5000
|
||||
|
||||
/**
|
||||
* How long after someone's last change they still count as typing.
|
||||
*
|
||||
* Long enough to ride out the pause between two words, short enough that the indicator means "right
|
||||
* now" rather than "recently". Only the two transitions are broadcast, not each keystroke.
|
||||
*/
|
||||
const TYPING_IDLE = 2000
|
||||
|
||||
/**
|
||||
* Cursor colours. Picked by hashing the user id, so one person is the same colour on everyone's screen
|
||||
* and stays that colour across sessions. Chosen to stay legible as a cursor label and as the
|
||||
* background of an avatar with white initials on it — hence no yellows or pastels.
|
||||
*/
|
||||
const USER_COLORS = [
|
||||
'#D32F2F',
|
||||
'#C2185B',
|
||||
'#7B1FA2',
|
||||
'#512DA8',
|
||||
'#303F9F',
|
||||
'#1976D2',
|
||||
'#0288D1',
|
||||
'#00796B',
|
||||
'#388E3C',
|
||||
'#E64A19',
|
||||
'#5D4037',
|
||||
'#455A64'
|
||||
]
|
||||
|
||||
let doc = null
|
||||
let provider = null
|
||||
let binding = null
|
||||
let styleEl = null
|
||||
let syncTimer = null
|
||||
/** Whether this author is mid-edit, and the timer that decides when they have stopped. */
|
||||
let typing = false
|
||||
let typingTimer = null
|
||||
/** Unsubscribe callbacks for the page store watchers, which have no component to be bound to. */
|
||||
let stopWatchers = []
|
||||
/**
|
||||
* Set while a remote change is being written into the page store, so the watcher that mirrors that
|
||||
* store back into the document does not send it round again.
|
||||
*/
|
||||
let applyingRemote = false
|
||||
|
||||
/**
|
||||
* A stable colour for a user.
|
||||
*
|
||||
* Exported because an avatar with no picture behind it is drawn in the same colour as its owner's
|
||||
* cursor — the whole point being that the face in the header and the caret in the text read as the
|
||||
* same person.
|
||||
*/
|
||||
export function collabUserColor(userId) {
|
||||
let hash = 0
|
||||
for (let index = 0; index < userId.length; index++) {
|
||||
hash = (hash * 31 + userId.charCodeAt(index)) | 0
|
||||
}
|
||||
return USER_COLORS[Math.abs(hash) % USER_COLORS.length]
|
||||
}
|
||||
|
||||
/** Whether a session is currently open. */
|
||||
export function isCollabActive() {
|
||||
return doc !== null
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a session on a page.
|
||||
*
|
||||
* Returns without waiting for the socket: the editor stays usable throughout, and the store's status
|
||||
* is what says whether anything is live yet.
|
||||
*/
|
||||
export function startCollabSession({ siteId, pageId }) {
|
||||
if (doc) {
|
||||
stopCollabSession()
|
||||
}
|
||||
|
||||
const collabStore = useCollabStore()
|
||||
const pageStore = usePageStore()
|
||||
const userStore = useUserStore()
|
||||
|
||||
doc = new Y.Doc()
|
||||
const ytext = doc.getText('content')
|
||||
const yprops = doc.getMap('props')
|
||||
const ymeta = doc.getMap('meta')
|
||||
|
||||
const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
|
||||
provider = new WebsocketProvider(
|
||||
`${protocol}//${window.location.host}/_collab`,
|
||||
`${siteId}/${pageId}`,
|
||||
doc
|
||||
)
|
||||
|
||||
collabStore.$patch({
|
||||
status: 'connecting',
|
||||
hasSynced: false,
|
||||
participants: [],
|
||||
lastSave: null
|
||||
})
|
||||
|
||||
provider.awareness.setLocalStateField('user', {
|
||||
id: userStore.id,
|
||||
name: userStore.name,
|
||||
hasAvatar: userStore.hasAvatar,
|
||||
color: collabUserColor(userStore.id)
|
||||
})
|
||||
|
||||
provider.awareness.on('change', refreshParticipants)
|
||||
|
||||
/*
|
||||
What makes an avatar pulse on everyone else's screen. `transaction.local` is the whole test: an
|
||||
edit this browser made is local, and one that arrived over the socket is not — so this fires for
|
||||
the author's own typing and never for the changes they are merely receiving. Header fields count
|
||||
too, being edits like any other.
|
||||
*/
|
||||
doc.on('update', (update, origin, updated, transaction) => {
|
||||
if (transaction?.local) {
|
||||
markTyping()
|
||||
}
|
||||
})
|
||||
|
||||
provider.on('status', ({ status }) => {
|
||||
/*
|
||||
`connected` here means the socket is up, which is not the same as the session being live — that
|
||||
is what `sync` below reports, and it is the only thing allowed to say `connected`. A refusal is
|
||||
final and outranks both.
|
||||
*/
|
||||
if (collabStore.status === 'denied' || status === 'connected') {
|
||||
return
|
||||
}
|
||||
collabStore.status = status === 'connecting' ? 'connecting' : 'disconnected'
|
||||
})
|
||||
|
||||
provider.on('sync', (isSynced) => {
|
||||
if (!isSynced) {
|
||||
return
|
||||
}
|
||||
clearTimeout(syncTimer)
|
||||
collabStore.$patch({ status: 'connected', hasSynced: true })
|
||||
/*
|
||||
The room may have been holding header fields somebody else changed and has not saved. Those are
|
||||
the current state of this edit, so they win over what this browser loaded from the API.
|
||||
*/
|
||||
adoptProps()
|
||||
refreshParticipants()
|
||||
})
|
||||
|
||||
provider.on('connection-close', (event) => {
|
||||
/*
|
||||
Codes in the 4000 range are the server's own (see `controllers/collab.ts`) and all mean the same
|
||||
thing: this session is not allowed, and reconnecting will be refused just as fast. Anything else
|
||||
is an ordinary drop, which the provider is right to retry.
|
||||
*/
|
||||
if (event?.code >= 4000) {
|
||||
collabStore.status = 'denied'
|
||||
provider.shouldConnect = false
|
||||
provider.disconnect()
|
||||
}
|
||||
})
|
||||
|
||||
/*
|
||||
Nothing is coming. A websocket that cannot be established — a proxy that does not forward upgrades
|
||||
is the usual reason — must not leave the author staring at an editor they are not allowed to type
|
||||
in, so the session gives up and the editor carries on as a plain one.
|
||||
*/
|
||||
syncTimer = setTimeout(() => {
|
||||
if (collabStore.status === 'connecting') {
|
||||
collabStore.status = 'disconnected'
|
||||
}
|
||||
}, SYNC_TIMEOUT)
|
||||
|
||||
// -> A header field somebody else edited, arriving mid-session
|
||||
yprops.observe((event, transaction) => {
|
||||
if (!transaction.local) {
|
||||
adoptProps()
|
||||
}
|
||||
})
|
||||
|
||||
// -> The server's word that the page has been saved. See `pageSaved` in `core/collab.ts`.
|
||||
ymeta.observe(() => {
|
||||
const info = ymeta.get('lastSave')
|
||||
if (info) {
|
||||
applySave(info)
|
||||
}
|
||||
})
|
||||
|
||||
/*
|
||||
The other direction: what this author types into the title, description or icon goes into the
|
||||
document. Watched on the store rather than bound to the inputs because those are three separate
|
||||
contenteditable elements in the page header, and the store is the one place all three meet.
|
||||
*/
|
||||
stopWatchers.push(
|
||||
watch(
|
||||
() => [pageStore.title, pageStore.description, pageStore.icon],
|
||||
([title, description, icon]) => {
|
||||
if (applyingRemote || !doc) {
|
||||
return
|
||||
}
|
||||
doc.transact(() => {
|
||||
writeProp(yprops, 'title', title)
|
||||
writeProp(yprops, 'description', description)
|
||||
writeProp(yprops, 'icon', icon)
|
||||
})
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
ensureStyleElement()
|
||||
|
||||
return { doc, ytext }
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand the Monaco model over to the session.
|
||||
*
|
||||
* Called once the document has synced, and not before: the binding starts by making the model say
|
||||
* what the document says, and a document that has not synced yet says nothing at all.
|
||||
*/
|
||||
export function bindCollabEditor(editor) {
|
||||
if (!doc || binding) {
|
||||
return
|
||||
}
|
||||
const model = editor.getModel()
|
||||
if (!model) {
|
||||
return
|
||||
}
|
||||
binding = new MonacoBinding(doc.getText('content'), model, new Set([editor]), provider.awareness)
|
||||
}
|
||||
|
||||
/** Close the session and put everything back the way an ordinary editor leaves it. */
|
||||
export function stopCollabSession() {
|
||||
clearTimeout(syncTimer)
|
||||
syncTimer = null
|
||||
clearTimeout(typingTimer)
|
||||
typingTimer = null
|
||||
typing = false
|
||||
for (const stop of stopWatchers) {
|
||||
stop()
|
||||
}
|
||||
stopWatchers = []
|
||||
if (binding) {
|
||||
binding.destroy()
|
||||
binding = null
|
||||
}
|
||||
if (provider) {
|
||||
// -> Retracts this editor's awareness state before the socket goes, so the others see the avatar
|
||||
// leave immediately rather than when the server notices the connection is gone
|
||||
provider.awareness.setLocalState(null)
|
||||
provider.destroy()
|
||||
provider = null
|
||||
}
|
||||
if (doc) {
|
||||
doc.destroy()
|
||||
doc = null
|
||||
}
|
||||
if (styleEl) {
|
||||
styleEl.remove()
|
||||
styleEl = null
|
||||
}
|
||||
applyingRemote = false
|
||||
useCollabStore().reset()
|
||||
}
|
||||
|
||||
// ----------------------------------------
|
||||
// Internals
|
||||
// ----------------------------------------
|
||||
|
||||
/**
|
||||
* Say that this author is typing, and arrange to say when they have stopped.
|
||||
*
|
||||
* Carried as an awareness field of its own rather than folded into `user`, so that a burst of typing
|
||||
* does not republish the name, colour and avatar with every change. Two messages per burst: one when
|
||||
* it starts, one when it ends.
|
||||
*/
|
||||
function markTyping() {
|
||||
if (!provider) {
|
||||
return
|
||||
}
|
||||
if (!typing) {
|
||||
typing = true
|
||||
provider.awareness.setLocalStateField('typing', true)
|
||||
}
|
||||
clearTimeout(typingTimer)
|
||||
typingTimer = setTimeout(() => {
|
||||
typing = false
|
||||
provider?.awareness.setLocalStateField('typing', false)
|
||||
}, TYPING_IDLE)
|
||||
}
|
||||
|
||||
function writeProp(yprops, key, value) {
|
||||
const next = value ?? ''
|
||||
if (yprops.get(key) !== next) {
|
||||
yprops.set(key, next)
|
||||
}
|
||||
}
|
||||
|
||||
/** Copy the shared header fields into the page store, without echoing them back out. */
|
||||
function adoptProps() {
|
||||
const pageStore = usePageStore()
|
||||
const yprops = doc.getMap('props')
|
||||
const patch = {}
|
||||
for (const key of ['title', 'description', 'icon']) {
|
||||
const value = yprops.get(key)
|
||||
// -> An icon is never legitimately empty, and blanking one because a room was seeded from a page
|
||||
// that had none would be a visible regression on every other screen
|
||||
if (typeof value !== 'string' || (key === 'icon' && !value)) {
|
||||
continue
|
||||
}
|
||||
if (value !== pageStore[key]) {
|
||||
patch[key] = value
|
||||
}
|
||||
}
|
||||
if (Object.keys(patch).length < 1) {
|
||||
return
|
||||
}
|
||||
applyingRemote = true
|
||||
pageStore.$patch(patch)
|
||||
// -> Released after the watchers have run, which they do synchronously only for `flush: 'sync'`
|
||||
// watchers; this one is deferred, so the flag has to outlive the tick
|
||||
queueMicrotask(() => {
|
||||
applyingRemote = false
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Somebody saved the page. Everyone else is now looking at what is stored, so their editor stops
|
||||
* claiming otherwise.
|
||||
*/
|
||||
function applySave(info) {
|
||||
const collabStore = useCollabStore()
|
||||
const editorStore = useEditorStore()
|
||||
const pageStore = usePageStore()
|
||||
|
||||
// -> The same instant in both, because "no pending changes" is those two fields being the same value
|
||||
const now = Temporal.Now.instant()
|
||||
editorStore.$patch({ lastChangeTimestamp: now, lastSaveTimestamp: now })
|
||||
pageStore.$patch({
|
||||
updatedAt: info.versionDate,
|
||||
authorId: info.authorId,
|
||||
authorName: info.authorName
|
||||
})
|
||||
collabStore.lastSave = info
|
||||
}
|
||||
|
||||
function refreshParticipants() {
|
||||
if (!provider || !doc) {
|
||||
return
|
||||
}
|
||||
const participants = []
|
||||
for (const [clientId, state] of provider.awareness.getStates()) {
|
||||
if (!state?.user?.id) {
|
||||
continue
|
||||
}
|
||||
participants.push({
|
||||
clientId,
|
||||
id: state.user.id,
|
||||
name: state.user.name || '',
|
||||
hasAvatar: Boolean(state.user.hasAvatar),
|
||||
color: state.user.color || collabUserColor(state.user.id),
|
||||
typing: Boolean(state.typing),
|
||||
isSelf: clientId === doc.clientID
|
||||
})
|
||||
}
|
||||
useCollabStore().participants = participants
|
||||
renderCursorStyles(participants)
|
||||
}
|
||||
|
||||
function ensureStyleElement() {
|
||||
if (styleEl) {
|
||||
return
|
||||
}
|
||||
styleEl = document.createElement('style')
|
||||
styleEl.dataset.collabCursors = 'true'
|
||||
document.head.appendChild(styleEl)
|
||||
}
|
||||
|
||||
/**
|
||||
* The stylesheet behind the remote cursors.
|
||||
*
|
||||
* y-monaco draws each remote selection as a decoration whose class carries the client id and nothing
|
||||
* else — `yRemoteSelection-42` — leaving what it looks like entirely to CSS. So one rule per
|
||||
* participant is generated here, which is also the only way the name can appear beside the caret: it
|
||||
* is drawn as generated content, there being no element to put it in.
|
||||
*/
|
||||
function renderCursorStyles(participants) {
|
||||
ensureStyleElement()
|
||||
styleEl.textContent = participants
|
||||
.filter((participant) => !participant.isSelf)
|
||||
.map(
|
||||
(participant) => `
|
||||
.yRemoteSelection-${participant.clientId} {
|
||||
background-color: ${participant.color}44;
|
||||
}
|
||||
.yRemoteSelectionHead-${participant.clientId} {
|
||||
position: relative;
|
||||
border-left: 2px solid ${participant.color};
|
||||
border-top: 2px solid ${participant.color};
|
||||
border-bottom: 2px solid ${participant.color};
|
||||
}
|
||||
.yRemoteSelectionHead-${participant.clientId}::after {
|
||||
content: '${cssString(participant.name)}';
|
||||
position: absolute;
|
||||
top: -1.4em;
|
||||
left: -2px;
|
||||
padding: 0 4px;
|
||||
border-radius: 2px 2px 2px 0;
|
||||
background-color: ${participant.color};
|
||||
color: #fff;
|
||||
font-size: 0.7rem;
|
||||
line-height: 1.4em;
|
||||
white-space: nowrap;
|
||||
pointer-events: none;
|
||||
user-select: none;
|
||||
}`
|
||||
)
|
||||
.join('\n')
|
||||
}
|
||||
|
||||
/** A user-supplied name, safe to sit inside a single-quoted CSS string. */
|
||||
function cssString(value) {
|
||||
return value
|
||||
.replace(/\\/g, '\\\\')
|
||||
.replace(/'/g, "\\'")
|
||||
.replace(/[\r\n]+/g, ' ')
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
import { defineStore } from 'pinia'
|
||||
|
||||
/**
|
||||
* Who else is editing the page that is open, and whether we are still hearing from them.
|
||||
*
|
||||
* The reactive face of `composables/collab.js`, which owns the Yjs document and the websocket and is
|
||||
* the only thing that writes here. Split off so that components — the avatars in the page header, the
|
||||
* editor itself — can read the session without touching any of that machinery.
|
||||
*/
|
||||
export const useCollabStore = defineStore('collab', {
|
||||
state: () => ({
|
||||
/**
|
||||
* - `off` — not collaborating: no session, or the site has the feature turned off
|
||||
* - `connecting` — the socket is up but the document has not been synced yet
|
||||
* - `connected` — live
|
||||
* - `disconnected` — the connection dropped and is being retried; edits are still safe locally
|
||||
* - `denied` — the server refused the session and retrying will not help
|
||||
*/
|
||||
status: 'off',
|
||||
/**
|
||||
* Whether the document has been synced at least once this session.
|
||||
*
|
||||
* A reconnection goes back through `connecting`, and the editor must not lock itself again over
|
||||
* it: by then there is a document in front of the author and edits made while the socket is down
|
||||
* are merged when it comes back. Only the very first sync is worth waiting for.
|
||||
*/
|
||||
hasSynced: false,
|
||||
/** One entry per open editor, so the same person in two tabs appears twice. */
|
||||
participants: [],
|
||||
/**
|
||||
* The last save anyone made to this page while the session has been open, as the server reported
|
||||
* it. Set only for a save that arrives during the session — the value a joining editor inherits
|
||||
* from the room is history, not news.
|
||||
*/
|
||||
lastSave: null
|
||||
}),
|
||||
getters: {
|
||||
isLive: (state) => state.status === 'connected',
|
||||
/**
|
||||
* One entry per person rather than per editor, which is what the header shows: two tabs are still
|
||||
* one face. Ordered with yourself first, as Google Docs and friends do.
|
||||
*/
|
||||
people: (state) => {
|
||||
const seen = new Map()
|
||||
for (const participant of state.participants) {
|
||||
const existing = seen.get(participant.id)
|
||||
if (existing) {
|
||||
// -> Two tabs are one face, and that face is typing if either of them is
|
||||
existing.typing = existing.typing || participant.typing
|
||||
} else {
|
||||
// -> Copied, since the merge above would otherwise write into the awareness snapshot
|
||||
seen.set(participant.id, { ...participant })
|
||||
}
|
||||
}
|
||||
return [...seen.values()].sort((a, b) => Number(b.isSelf) - Number(a.isSelf))
|
||||
}
|
||||
},
|
||||
actions: {
|
||||
reset() {
|
||||
this.$patch({ status: 'off', hasSynced: false, participants: [], lastSave: null })
|
||||
}
|
||||
}
|
||||
})
|
||||
Loading…
Reference in new issue