// OT example: Two users edit "Hello"
// User A: Insert "!" at position 5 → "Hello!"
// User B: Insert " World" at position 5 → "Hello World"

// Server must transform operations:
// - Apply A's operation first
// - Transform B's operation: position 5 → position 6 (shifted by "!")
// - Result: "Hello! World"

const Y = require('yjs');

// Create a shared document
const doc = new Y.Doc();

// Create shared types
const yText = doc.getText('content');
const yMap = doc.getMap('metadata');
const yArray = doc.getArray('todos');

// Listen for changes
doc.on('update', (update, origin) => {
  console.log('Document updated:', update);
  // Send update to other clients via WebSocket
});

// Apply remote updates
const remoteUpdate = new Uint8Array([/* binary CRDT update */]);
Y.applyUpdate(doc, remoteUpdate);

// User A's doc:
const docA = new Y.Doc();
const textA = docA.getText('content');
textA.insert(0, 'Hello');

// User B's doc (independent):
const docB = new Y.Doc();
const textB = docB.getText('content');
textB.insert(0, 'World');

// Sync: Each character has a unique ID
// A's doc: [H:A1, e:A2, l:A3, l:A4, o:A5]
// B's doc: [W:B1, o:B2, r:B3, l:B4, d:B5]

// Merge (order by ID):
// [H:A1, W:B1, e:A2, o:B2, l:A3, r:B3, l:A4, l:B4, o:A5, d:B5]
// Result: "HWeolrllod" ← Deterministic but nonsensical

// Fix: Yjs uses fractional indexing for sensible merges
// Actual result: "HelloWorld" or "WorldHello" (deterministic per algorithm)

const WebSocket = require('ws');
const Y = require('yjs');
const { setupWSConnection } = require('y-websocket/bin/utils');

const wss = new WebSocket.Server({ port: 1234 });

// Store in-memory documents (use Redis/Postgres for persistence)
const docs = new Map();

wss.on('connection', (ws, req) => {
  // Extract document ID from URL: /documents/doc-123
  const docId = new URL(req.url, 'http://localhost').pathname.split('/')[2];

  if (!docId) {
    ws.close();
    return;
  }

  // Get or create shared document
  if (!docs.has(docId)) {
    const doc = new Y.Doc();
    docs.set(docId, doc);

    // Optional: Load from database
    loadDocumentFromDB(docId).then(state => {
      if (state) {
        Y.applyUpdate(doc, state);
      }
    });
  }

  const doc = docs.get(docId);

  // Use y-websocket's built-in sync protocol
  setupWSConnection(ws, req, { doc });

  console.log(`Client connected to document: ${docId}`);
});

// Persist documents every 30 seconds
setInterval(() => {
  docs.forEach((doc, docId) => {
    const state = Y.encodeStateAsUpdate(doc);
    saveDocumentToDB(docId, state);
  });
}, 30000);

import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';
import { CodemirrorBinding } from 'y-codemirror';
import CodeMirror from 'codemirror';

// Initialize Yjs document
const doc = new Y.Doc();

// Connect to WebSocket server
const provider = new WebsocketProvider(
  'ws://localhost:1234', 
  'doc-123', // Document ID
  doc
);

provider.on('status', event => {
  console.log(event.status); // 'connected', 'disconnected'
});

// Bind to CodeMirror editor
const yText = doc.getText('content');
const editor = CodeMirror(document.querySelector('#editor'), {
  mode: 'javascript',
  lineNumbers: true
});

const binding = new CodemirrorBinding(yText, editor, provider.awareness);

// Now typing in CodeMirror automatically syncs to other clients!

const Redis = require('ioredis');
const subscriber = new Redis();
const publisher = new Redis();

const rooms = new Map(); // docId → Set<WebSocket>

subscriber.on('message', (channel, message) => {
  const [docId, update] = JSON.parse(message);
  const clients = rooms.get(docId);

  if (clients) {
    const buffer = Buffer.from(update, 'base64');
    clients.forEach(ws => {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(buffer);
      }
    });
  }
});

wss.on('connection', (ws, req) => {
  const docId = extractDocId(req.url);

  // Join room
  if (!rooms.has(docId)) {
    rooms.set(docId, new Set());
    subscriber.subscribe(`doc:${docId}`);
  }
  rooms.get(docId).add(ws);

  // Broadcast updates
  ws.on('message', data => {
    // Publish to Redis (other servers receive it)
    publisher.publish(`doc:${docId}`, JSON.stringify([
      docId,
      data.toString('base64')
    ]));
  });

  ws.on('close', () => {
    const clients = rooms.get(docId);
    clients.delete(ws);

    if (clients.size === 0) {
      rooms.delete(docId);
      subscriber.unsubscribe(`doc:${docId}`);
    }
  });
});

const { setupWSConnection } = require('y-websocket/bin/utils');

wss.on('connection', (ws, req) => {
  const doc = docs.get(docId);

  // setupWSConnection handles both document sync AND awareness
  setupWSConnection(ws, req, { doc });

  // Awareness state is automatically broadcast to all clients
});

const provider = new WebsocketProvider('ws://localhost:1234', 'doc-123', doc);

// Set local user info
provider.awareness.setLocalStateField('user', {
  name: 'Alice',
  color: '#ff6b6b',
  cursor: { line: 10, ch: 5 }
});

// Listen for other users' cursors
provider.awareness.on('change', changes => {
  changes.added.forEach(clientId => {
    const user = provider.awareness.getStates().get(clientId).user;
    console.log(`${user.name} joined`);
    renderRemoteCursor(user);
  });

  changes.updated.forEach(clientId => {
    const user = provider.awareness.getStates().get(clientId).user;
    updateRemoteCursor(user);
  });

  changes.removed.forEach(clientId => {
    removeRemoteCursor(clientId);
  });
});

function renderRemoteCursor(user) {
  const cursorEl = document.createElement('div');
  cursorEl.className = 'remote-cursor';
  cursorEl.style.backgroundColor = user.color;
  cursorEl.textContent = user.name;
  // Position at user.cursor coordinates
  editor.addWidget(user.cursor, cursorEl);
}

CREATE TABLE documents (
  id TEXT PRIMARY KEY,
  state BYTEA NOT NULL,
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_documents_updated ON documents(updated_at);

const { Pool } = require('pg');
const pool = new Pool();

async function saveDocumentToDB(docId, state) {
  await pool.query(
    `INSERT INTO documents (id, state, updated_at) 
     VALUES ($1, $2, NOW())
     ON CONFLICT (id) DO UPDATE 
     SET state = $2, updated_at = NOW()`,
    [docId, Buffer.from(state)]
  );
}

async function loadDocumentFromDB(docId) {
  const { rows } = await pool.query(
    'SELECT state FROM documents WHERE id = $1',
    [docId]
  );
  return rows[0]?.state;
}

CREATE TABLE document_updates (
  id BIGSERIAL PRIMARY KEY,
  document_id TEXT NOT NULL,
  update BYTEA NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_updates_doc ON document_updates(document_id, created_at);

async function loadDocumentFromDB(docId) {
  const { rows } = await pool.query(
    'SELECT update FROM document_updates WHERE document_id = $1 ORDER BY created_at',
    [docId]
  );

  const doc = new Y.Doc();
  rows.forEach(row => {
    Y.applyUpdate(doc, row.update);
  });

  return Y.encodeStateAsUpdate(doc);
}

doc.on('update', async (update, origin) => {
  if (origin !== 'db') { // Avoid saving updates from DB load
    await pool.query(
      'INSERT INTO document_updates (document_id, update) VALUES ($1, $2)',
      [docId, Buffer.from(update)]
    );
  }
});

// Periodic compaction: Merge old updates into snapshot
setInterval(async () => {
  const cutoff = new Date(Date.now() - 24 * 3600 * 1000); // 1 day ago

  const { rows } = await pool.query(
    'SELECT update FROM document_updates WHERE document_id = $1 AND created_at < $2',
    [docId, cutoff]
  );

  if (rows.length > 100) {
    const doc = new Y.Doc();
    rows.forEach(row => Y.applyUpdate(doc, row.update));
    const snapshot = Y.encodeStateAsUpdate(doc);

    await pool.query('BEGIN');
    await pool.query(
      'INSERT INTO documents (id, state) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET state = $2',
      [docId, Buffer.from(snapshot)]
    );
    await pool.query(
      'DELETE FROM document_updates WHERE document_id = $1 AND created_at < $2',
      [docId, cutoff]
    );
    await pool.query('COMMIT');
  }
}, 3600000); // Every hour

const Redis = require('ioredis');
const redis = new Redis();

async function saveDocumentToRedis(docId, state) {
  await redis.setex(
    `doc:${docId}`,
    3600, // 1 hour TTL
    Buffer.from(state).toString('base64')
  );
}

async function loadDocumentFromRedis(docId) {
  const data = await redis.get(`doc:${docId}`);
  return data ? Buffer.from(data, 'base64') : null;
}

// Fallback to DB if not in Redis
async function loadDocument(docId) {
  let state = await loadDocumentFromRedis(docId);
  if (!state) {
    state = await loadDocumentFromDB(docId);
    if (state) {
      await saveDocumentToRedis(docId, state);
    }
  }
  return state;
}

// Initial state: "Hello"

// User A: Insert " World" at position 5 → "Hello World"
const docA = new Y.Doc();
const textA = docA.getText('content');
textA.insert(0, 'Hello');
textA.insert(5, ' World');

// User B (simultaneously): Insert "!" at position 5 → "Hello!"
const docB = new Y.Doc();
const textB = docB.getText('content');
textB.insert(0, 'Hello');
textB.insert(5, '!');

// Sync updates
const updateA = Y.encodeStateAsUpdate(docA);
const updateB = Y.encodeStateAsUpdate(docB);

Y.applyUpdate(docA, updateB);
Y.applyUpdate(docB, updateA);

console.log(textA.toString()); // "Hello World!" (deterministic)
console.log(textB.toString()); // "Hello World!" (same)

const docA = new Y.Doc();
const mapA = docA.getMap('config');
mapA.set('theme', 'dark');

const docB = new Y.Doc();
const mapB = docB.getMap('config');
mapB.set('theme', 'light');

// Sync
Y.applyUpdate(docA, Y.encodeStateAsUpdate(docB));
Y.applyUpdate(docB, Y.encodeStateAsUpdate(docA));

// Result: Last-write-wins (based on vector clock timestamps)
console.log(mapA.get('theme')); // "light" or "dark" (deterministic per timestamp)

const docA = new Y.Doc();
const arrayA = docA.getArray('todos');
arrayA.push(['Task 1']);

const docB = new Y.Doc();
const arrayB = docB.getArray('todos');
arrayB.push(['Task 2']);

// Sync
Y.applyUpdate(docA, Y.encodeStateAsUpdate(docB));

console.log(arrayA.toArray()); // ['Task 1', 'Task 2'] (deterministic order)

// Compact document by removing tombstones
const compactedState = Y.encodeStateAsUpdate(doc);
const newDoc = new Y.Doc();
Y.applyUpdate(newDoc, compactedState);

// Replace old doc with compacted version
doc = newDoc;

const jwt = require('jsonwebtoken');

wss.on('connection', (ws, req) => {
  const token = new URL(req.url, 'http://localhost').searchParams.get('token');

  try {
    const payload = jwt.verify(token, process.env.JWT_SECRET);
    const docId = extractDocId(req.url);

    // Check if user has access to document
    if (!userHasAccess(payload.userId, docId)) {
      ws.close(4003, 'Forbidden');
      return;
    }

    ws.userId = payload.userId;
    setupWSConnection(ws, req, { doc: docs.get(docId) });
  } catch (err) {
    ws.close(4001, 'Unauthorized');
  }
});

const rateLimits = new Map(); // userId → { count, resetAt }

ws.on('message', data => {
  const limit = rateLimits.get(ws.userId) || { count: 0, resetAt: Date.now() + 1000 };

  if (Date.now() > limit.resetAt) {
    limit.count = 0;
    limit.resetAt = Date.now() + 1000;
  }

  if (limit.count >= 100) {
    ws.send(JSON.stringify({ error: 'Rate limit exceeded' }));
    return;
  }

  limit.count++;
  rateLimits.set(ws.userId, limit);

  // Process update
  broadcastUpdate(data);
});

const prometheus = require('prom-client');

const activeConnections = new prometheus.Gauge({
  name: 'websocket_active_connections',
  help: 'Number of active WebSocket connections',
  labelNames: ['document_id']
});

const updateSize = new prometheus.Histogram({
  name: 'yjs_update_size_bytes',
  help: 'Size of Yjs updates',
  buckets: [10, 50, 100, 500, 1000, 5000]
});

wss.on('connection', (ws, req) => {
  const docId = extractDocId(req.url);
  activeConnections.inc({ document_id: docId });

  ws.on('message', data => {
    updateSize.observe(data.length);
  });

  ws.on('close', () => {
    activeConnections.dec({ document_id: docId });
  });
});

const doc = new Y.Doc();
const xmlFragment = doc.getXmlFragment('prosemirror');

// Bind to ProseMirror
const binding = new ProsemirrorBinding(xmlFragment, view);

const doc = new Y.Doc();
const version = doc.getMap('_meta').get('version') || 1;

if (version < 2) {
  // Migrate from v1 to v2
  const oldData = doc.getMap('data');
  const newData = doc.getMap('data_v2');

  oldData.forEach((value, key) => {
    newData.set(key, transformV1toV2(value));
  });

  doc.getMap('_meta').set('version', 2);
}

const docA = new Y.Doc();
const textA = docA.getText('content');
textA.insert(0, 'Hello');
textA.delete(1, 4); // Delete "ello" → "H"

const docB = new Y.Doc();
const textB = docB.getText('content');
textB.insert(0, 'Hello');
textB.delete(0, 5); // Delete entire text → ""

// Sync
Y.applyUpdate(docA, Y.encodeStateAsUpdate(docB));
console.log(textA.toString()); // "" (both deletes applied)

const undoManager = new Y.UndoManager(yText);

// User types
yText.insert(0, 'Hello');
yText.insert(5, ' World');

// Undo
undoManager.undo(); // Removes " World"
console.log(yText.toString()); // "Hello"

// Redo
undoManager.redo(); // Restores " World"
console.log(yText.toString()); // "Hello World"

// GraphQL subscription
subscription {
  documentUpdates(docId: "doc-123") {
    update # Base64-encoded Yjs update
  }
}

// Client
apolloClient.subscribe({
  query: DOCUMENT_UPDATES,
  variables: { docId: 'doc-123' }
}).subscribe(({ data }) => {
  const update = Buffer.from(data.documentUpdates.update, 'base64');
  Y.applyUpdate(doc, update);
});