SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Replace Durable Object relay with Trystero P2P + y-indexeddb

Browse source 
- Remove packages/collab/ (CF Worker + Durable Object server)
- Replace WebSocket relay with Trystero (serverless WebRTC P2P)
  - Peers connect via Nostr relays for signaling (free, public)
  - All sync traffic flows directly peer-to-peer
  - Automatic chunking for large Yjs updates
- Add y-indexeddb for local persistence
  - Each client persists Y.Doc to IndexedDB
  - Room survives browser refresh, loads instantly
- Yjs sync protocol over Trystero actions:
  - yjs-update: incremental doc updates
  - sync-step1/sync-reply: full state sync on peer join
  - awareness: cursor, selection, presence
- Zero server cost at any scale
This commit is contained in:
Danila Poyarkov 2026-03-01 17:25:46 +03:00
parent 8695ba6d1b
commit 3427b7fd3c
7 changed files with 628 additions and 471 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

636
bun.lock
View file

File diff suppressed because it is too large Load diff

2
package.json
View file

@ -51,10 +51,12 @@
"prismjs": "^1.30.0",
"reka-ui": "^2.8.2",
"tailwindcss": "^4.2.1",
"trystero": "^0.22.0",
"valibot": "^1.2.0",
"vue": "^3.5.29",
"vue-router": "^5.0.3",
"vue-stream-markdown": "^0.6.0",
"y-indexeddb": "^9.0.12",
"y-protocols": "^1.0.7",
"yjs": "^13.6.29",
"yoga-layout": "^3.2.1"

18
packages/collab/package.json
View file

@ -1,18 +0,0 @@
{
"name": "@open-pencil/collab",
"version": "0.1.0",
"private": true,
"scripts": {
"dev": "wrangler dev",
"deploy": "wrangler deploy",
"types": "wrangler types"
},
"devDependencies": {
"@cloudflare/workers-types": "^4.20250214.0",
"wrangler": "^4"
},
"dependencies": {
"yjs": "^13.6.24",
"y-protocols": "^1.0.6"
}
}

204
packages/collab/src/index.ts
View file

@ -1,204 +0,0 @@
import { DurableObject } from 'cloudflare:workers'
import * as Y from 'yjs'
import * as syncProtocol from 'y-protocols/sync'
import * as awarenessProtocol from 'y-protocols/awareness'
import * as encoding from 'lib0/encoding'
import * as decoding from 'lib0/decoding'
const MSG_SYNC = 0
const MSG_AWARENESS = 1
interface Env {
ROOMS: DurableObjectNamespace<CollabRoom>
}
export class CollabRoom extends DurableObject {
doc: Y.Doc
awareness: awarenessProtocol.Awareness
constructor(ctx: DurableObjectState, env: Env) {
super(ctx, env)
this.doc = new Y.Doc()
this.awareness = new awarenessProtocol.Awareness(this.doc)
// Restore persisted doc state
this.ctx.blockConcurrencyWhile(async () => {
const stored = await this.ctx.storage.get<Uint8Array>('doc')
if (stored) {
Y.applyUpdate(this.doc, new Uint8Array(stored))
}
})
// Restore awareness from hibernated WebSockets
for (const ws of this.ctx.getWebSockets()) {
const tag = ws.deserializeAttachment() as { clientId: number } | null
if (tag) {
// Awareness state will be re-sent by the client on reconnect
}
}
this.doc.on('update', (_update: Uint8Array, origin: unknown) => {
if (origin === 'remote') return
// Persist on server-originated updates (rare, mostly initial)
this.persistDoc()
})
}
async fetch(request: Request): Promise<Response> {
const url = new URL(request.url)
if (url.pathname === '/health') {
return new Response('ok')
}
const upgradeHeader = request.headers.get('Upgrade')
if (!upgradeHeader || upgradeHeader !== 'websocket') {
return new Response('Expected WebSocket', { status: 426 })
}
const pair = new WebSocketPair()
const [client, server] = Object.values(pair)
this.ctx.acceptWebSocket(server)
const clientId = Math.floor(Math.random() * 0xFFFFFF)
server.serializeAttachment({ clientId })
// Send sync step 1 to the new peer
const syncEncoder = encoding.createEncoder()
encoding.writeVarUint(syncEncoder, MSG_SYNC)
syncProtocol.writeSyncStep1(syncEncoder, this.doc)
server.send(encoding.toUint8Array(syncEncoder))
// Send current awareness states
const awarenessStates = this.awareness.getStates()
if (awarenessStates.size > 0) {
const awarenessEncoder = encoding.createEncoder()
encoding.writeVarUint(awarenessEncoder, MSG_AWARENESS)
const update = awarenessProtocol.encodeAwarenessUpdate(
this.awareness,
Array.from(awarenessStates.keys())
)
encoding.writeVarUint8Array(awarenessEncoder, update)
server.send(encoding.toUint8Array(awarenessEncoder))
}
return new Response(null, { status: 101, webSocket: client })
}
async webSocketMessage(ws: WebSocket, message: ArrayBuffer | string) {
if (typeof message === 'string') return
const data = new Uint8Array(message)
const decoder = decoding.createDecoder(data)
const msgType = decoding.readVarUint(decoder)
switch (msgType) {
case MSG_SYNC: {
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, MSG_SYNC)
const syncMessageType = syncProtocol.readSyncMessage(decoder, encoder, this.doc, 'remote')
if (encoding.length(encoder) > 1) {
ws.send(encoding.toUint8Array(encoder))
}
// If we received an update (step 2 or update), broadcast to others
if (syncMessageType === 1 || syncMessageType === 2) {
this.broadcastExcept(ws, data)
this.persistDoc()
}
break
}
case MSG_AWARENESS: {
const update = decoding.readVarUint8Array(decoder)
awarenessProtocol.applyAwarenessUpdate(this.awareness, update, ws)
this.broadcastExcept(ws, data)
break
}
}
}
async webSocketClose(ws: WebSocket, code: number, reason: string) {
const tag = ws.deserializeAttachment() as { clientId: number } | null
if (tag) {
awarenessProtocol.removeAwarenessStates(this.awareness, [tag.clientId], 'peer left')
// Broadcast awareness removal
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, MSG_AWARENESS)
const update = awarenessProtocol.encodeAwarenessUpdate(this.awareness, [tag.clientId])
encoding.writeVarUint8Array(encoder, update)
this.broadcastExcept(ws, encoding.toUint8Array(encoder))
}
ws.close(code, reason)
}
async webSocketError(ws: WebSocket) {
const tag = ws.deserializeAttachment() as { clientId: number } | null
if (tag) {
awarenessProtocol.removeAwarenessStates(this.awareness, [tag.clientId], 'error')
}
}
private broadcastExcept(sender: WebSocket, data: Uint8Array) {
for (const ws of this.ctx.getWebSockets()) {
if (ws !== sender) {
try {
ws.send(data)
} catch {
// Peer disconnected
}
}
}
}
private async persistDoc() {
const state = Y.encodeStateAsUpdate(this.doc)
await this.ctx.storage.put('doc', state)
}
}
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const url = new URL(request.url)
// CORS preflight
if (request.method === 'OPTIONS') {
return new Response(null, {
headers: corsHeaders()
})
}
// Route: /room/:roomId
const match = url.pathname.match(/^\/room\/([a-zA-Z0-9_-]+)$/)
if (!match) {
return new Response('Not found', { status: 404, headers: corsHeaders() })
}
const roomId = match[1]
const id = env.ROOMS.idFromName(roomId)
const stub = env.ROOMS.get(id)
const response = await stub.fetch(request)
// Add CORS headers for non-WebSocket responses
if (response.status !== 101) {
const headers = new Headers(response.headers)
for (const [key, value] of Object.entries(corsHeaders())) {
headers.set(key, value)
}
return new Response(response.body, { status: response.status, headers })
}
return response
}
} satisfies ExportedHandler<Env>
function corsHeaders(): Record<string, string> {
return {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, OPTIONS',
'Access-Control-Allow-Headers': 'Upgrade, Content-Type'
}
}

13
packages/collab/tsconfig.json
View file

@ -1,13 +0,0 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "ES2022",
"moduleResolution": "bundler",
"lib": ["ES2022"],
"types": ["@cloudflare/workers-types/2023-07-01"],
"strict": true,
"noEmit": true,
"skipLibCheck": true
},
"include": ["src/**/*.ts"]
}

26
packages/collab/wrangler.jsonc
View file

@ -1,26 +0,0 @@
{
"$schema": "node_modules/wrangler/config-schema.json",
"name": "open-pencil-collab",
"main": "src/index.ts",
"compatibility_date": "2025-02-26",
"durable_objects": {
"bindings": [
{
"name": "ROOMS",
"class_name": "CollabRoom"
}
]
},
"migrations": [
{
"tag": "v1",
"new_sqlite_classes": ["CollabRoom"]
}
],
"routes": [
{
"pattern": "collab.openpencil.dev",
"custom_domain": true
}
]
}

200
src/composables/use-collab.ts
View file

@ -1,18 +1,15 @@
import * as decoding from 'lib0/decoding'
import * as encoding from 'lib0/encoding'
import { ref, onUnmounted, computed } from 'vue'
import * as awarenessProtocol from 'y-protocols/awareness'
import * as syncProtocol from 'y-protocols/sync'
import { IndexeddbPersistence } from 'y-indexeddb'
import { joinRoom as joinTrysteroRoom } from 'trystero'
import type { Room } from 'trystero'
import * as Y from 'yjs'
import type { SceneNode } from '@/engine/scene-graph'
import type { EditorStore } from '@/stores/editor'
import type { Color } from '@/types'
const MSG_SYNC = 0
const MSG_AWARENESS = 1
const COLLAB_URL = import.meta.env.VITE_COLLAB_URL || 'wss://collab.openpencil.dev'
const TRYSTERO_APP_ID = 'openpencil'
const PEER_COLORS: Color[] = [
{ r: 0.96, g: 0.26, b: 0.21, a: 1 },
@ -22,7 +19,7 @@ const PEER_COLORS: Color[] = [
{ r: 0.61, g: 0.15, b: 0.69, a: 1 },
{ r: 1.0, g: 0.34, b: 0.13, a: 1 },
{ r: 0.0, g: 0.74, b: 0.83, a: 1 },
{ r: 0.91, g: 0.12, b: 0.39, a: 1 }
{ r: 0.91, g: 0.12, b: 0.39, a: 1 },
]
export interface RemotePeer {
@ -47,34 +44,36 @@ export function useCollab(store: EditorStore) {
roomId: null,
peers: [],
localName: localStorage.getItem('op-collab-name') || '',
localColor: PEER_COLORS[Math.floor(Math.random() * PEER_COLORS.length)]
localColor: PEER_COLORS[Math.floor(Math.random() * PEER_COLORS.length)],
})
let ws: WebSocket | null = null
let ydoc: Y.Doc | null = null
let awareness: awarenessProtocol.Awareness | null = null
let ynodes: Y.Map<Y.Map<unknown>> | null = null
let ymeta: Y.Map<unknown> | null = null
let room: Room | null = null
let persistence: IndexeddbPersistence | null = null
let suppressGraphEvents = false
let suppressYjsEvents = false
let sendYjsUpdate: ((data: Uint8Array, peerId?: string) => void) | null = null
let sendAwareness: ((data: Uint8Array, peerId?: string) => void) | null = null
let sendSyncStep1: ((data: Uint8Array, peerId?: string) => void) | null = null
const remotePeers = computed(() => state.value.peers)
function connect(roomId: string) {
if (ws) disconnect()
if (room) disconnect()
state.value.roomId = roomId
ydoc = new Y.Doc()
awareness = new awarenessProtocol.Awareness(ydoc)
ynodes = ydoc.getMap('nodes')
ymeta = ydoc.getMap('meta')
// Listen for awareness changes → update peers list
persistence = new IndexeddbPersistence(`op-room-${roomId}`, ydoc)
awareness.on('change', () => {
updatePeersList()
})
// Listen for remote Yjs changes → apply to SceneGraph
ynodes.observeDeep((events) => {
if (suppressYjsEvents) return
suppressGraphEvents = true
@ -86,41 +85,78 @@ export function useCollab(store: EditorStore) {
store.requestRender()
})
// Register doc update listener before opening WebSocket
ydoc.on('update', (update: Uint8Array, origin: unknown) => {
if (origin === 'remote') return
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, MSG_SYNC)
syncProtocol.writeUpdate(encoder, update)
sendBinary(encoding.toUint8Array(encoder))
room = joinTrysteroRoom({ appId: TRYSTERO_APP_ID }, roomId)
const [sendUpdate, getUpdate] = room.makeAction<Uint8Array>('yjs-update')
const [sendAw, getAw] = room.makeAction<Uint8Array>('awareness')
const [sendSync, getSync] = room.makeAction<Uint8Array>('sync-step1')
const [sendSyncReply, getSyncReply] = room.makeAction<Uint8Array>('sync-reply')
sendYjsUpdate = (data, peerId) =>
peerId ? sendUpdate(data, peerId) : sendUpdate(data)
sendAwareness = (data, peerId) =>
peerId ? sendAw(data, peerId) : sendAw(data)
sendSyncStep1 = (data, peerId) =>
peerId ? sendSync(data, peerId) : sendSync(data)
getUpdate((data) => {
if (!ydoc) return
Y.applyUpdate(ydoc, new Uint8Array(data), 'remote')
})
// WebSocket connection
const url = `${COLLAB_URL}/room/${roomId}`
ws = new WebSocket(url)
ws.binaryType = 'arraybuffer'
getAw((data) => {
if (!awareness) return
awarenessProtocol.applyAwarenessUpdate(awareness, new Uint8Array(data), null)
})
ws.onopen = () => {
getSync((data, peerId) => {
if (!ydoc) return
const sv = new Uint8Array(data)
const update = Y.encodeStateAsUpdate(ydoc, sv)
sendSyncReply(update, peerId)
})
getSyncReply((data) => {
if (!ydoc) return
Y.applyUpdate(ydoc, new Uint8Array(data), 'remote')
})
ydoc.on('update', (update: Uint8Array, origin: unknown) => {
if (origin === 'remote') return
sendYjsUpdate?.(update)
})
awareness.on('update', ({ added, updated, removed }: {
added: number[]
updated: number[]
removed: number[]
}) => {
const changedClients = [...added, ...updated, ...removed]
const encodedUpdate = awarenessProtocol.encodeAwarenessUpdate(awareness!, changedClients)
sendAwareness?.(encodedUpdate)
})
room.onPeerJoin((peerId) => {
state.value.connected = true
broadcastAwareness()
if (ymeta) ymeta.set('roomId', roomId)
}
const sv = Y.encodeStateVector(ydoc!)
sendSyncStep1?.(sv, peerId)
ws.onmessage = (event) => {
if (!(event.data instanceof ArrayBuffer)) return
handleMessage(new Uint8Array(event.data))
}
if (awareness) {
const encodedUpdate = awarenessProtocol.encodeAwarenessUpdate(
awareness,
[awareness.clientID]
)
sendAwareness?.(encodedUpdate, peerId)
}
})
ws.onclose = () => {
state.value.connected = false
// TODO: reconnect logic
}
room.onPeerLeave(() => {
updatePeersList()
})
ws.onerror = () => {
state.value.connected = false
}
state.value.connected = true
broadcastAwareness()
// Sync local SceneGraph → Yjs on graph mutations
const origUpdateNode = store.graph.updateNode.bind(store.graph)
store.graph.updateNode = (id: string, changes: Partial<SceneNode>) => {
origUpdateNode(id, changes)
@ -131,59 +167,32 @@ export function useCollab(store: EditorStore) {
}
function disconnect() {
if (ws) {
ws.close()
ws = null
}
room?.leave()
room = null
sendYjsUpdate = null
sendAwareness = null
sendSyncStep1 = null
if (awareness) {
awareness.destroy()
awareness = null
}
if (persistence) {
persistence.destroy()
persistence = null
}
if (ydoc) {
ydoc.destroy()
ydoc = null
}
ynodes = null
ymeta = null
state.value.connected = false
state.value.roomId = null
state.value.peers = []
store.state.remoteCursors = []
store.requestRender()
}
function handleMessage(data: Uint8Array) {
if (!ydoc || !awareness) return
const decoder = decoding.createDecoder(data)
const msgType = decoding.readVarUint(decoder)
switch (msgType) {
case MSG_SYNC: {
const encoder = encoding.createEncoder()
encoding.writeVarUint(encoder, MSG_SYNC)
syncProtocol.readSyncMessage(decoder, encoder, ydoc, null)
if (encoding.length(encoder) > 1) {
sendBinary(encoding.toUint8Array(encoder))
}
break
}
case MSG_AWARENESS: {
const update = decoding.readVarUint8Array(decoder)
awarenessProtocol.applyAwarenessUpdate(awareness, update, null)
break
}
}
}
function sendBinary(data: Uint8Array) {
if (ws && ws.readyState === WebSocket.OPEN) {
ws.send(data)
}
}
// Sync doc updates to server
function syncNodeToYjs(nodeId: string) {
if (!ydoc || !ynodes) return
const node = store.graph.getNode(nodeId)
@ -230,7 +239,6 @@ export function useCollab(store: EditorStore) {
function applyYjsToGraph(events: Y.YEvent<Y.Map<unknown>>[]) {
for (const event of events) {
if (event.target === ynodes) {
// Top-level additions/deletions of nodes
for (const [key, change] of event.changes.keys) {
if (change.action === 'add') {
const ynode = ynodes!.get(key)
@ -240,7 +248,6 @@ export function useCollab(store: EditorStore) {
}
}
} else if (event.target.parent === ynodes) {
// Property changes within a node's Y.Map
const nodeId = findNodeIdForYMap(event.target as Y.Map<unknown>)
if (nodeId) {
const ynode = ynodes!.get(nodeId)
@ -259,19 +266,15 @@ export function useCollab(store: EditorStore) {
}
function applyYnodeToGraph(nodeId: string, ynode: Y.Map<unknown>) {
const JSON_FIELDS = new Set([
'childIds', 'fills', 'strokes', 'effects',
'vectorNetwork', 'boundVariables', 'styleRuns',
])
const existing = store.graph.getNode(nodeId)
const props: Record<string, unknown> = {}
for (const [key, value] of ynode.entries()) {
if (
key === 'childIds' ||
key === 'fills' ||
key === 'strokes' ||
key === 'effects' ||
key === 'vectorNetwork' ||
key === 'boundVariables' ||
key === 'styleRuns'
) {
if (JSON_FIELDS.has(key)) {
try {
props[key] = typeof value === 'string' ? JSON.parse(value) : value
} catch {
@ -285,12 +288,10 @@ export function useCollab(store: EditorStore) {
if (existing) {
store.graph.updateNode(nodeId, props as Partial<SceneNode>)
} else {
// New node from remote — create it
const parentId = props.parentId as string
if (parentId && store.graph.getNode(parentId)) {
const type = props.type as SceneNode['type']
const node = store.graph.createNode(type, parentId, props as Partial<SceneNode>)
// Override the auto-generated id with the actual id
store.graph.nodes.delete(node.id)
node.id = nodeId
store.graph.nodes.set(nodeId, node)
@ -298,12 +299,11 @@ export function useCollab(store: EditorStore) {
}
}
// Awareness: broadcast local cursor/selection
function broadcastAwareness() {
if (!awareness) return
awareness.setLocalStateField('user', {
name: state.value.localName,
color: state.value.localColor
color: state.value.localColor,
})
}
@ -333,13 +333,11 @@ export function useCollab(store: EditorStore) {
name: user.name || 'Anonymous',
color: user.color || PEER_COLORS[clientId % PEER_COLORS.length],
cursor: peerState.cursor as RemotePeer['cursor'],
selection: peerState.selection as string[]
selection: peerState.selection as string[],
})
})
state.value.peers = peers
// Update store's remoteCursors for renderer
store.state.remoteCursors = peers
.filter((p) => p.cursor && p.cursor.pageId === currentPageId)
.map((p) => ({
@ -347,7 +345,7 @@ export function useCollab(store: EditorStore) {
color: p.color,
x: p.cursor!.x,
y: p.cursor!.y,
selection: p.selection
selection: p.selection,
}))
store.requestRender()
}
@ -390,6 +388,6 @@ export function useCollab(store: EditorStore) {
shareCurrentDoc,
updateCursor,
updateSelection,
setLocalName
setLocalName,
}
}