migrate from github

.dockerignore

@@ -0,0 +1,11 @@
+node_modules/
+dist/
+.git/
+.vscode/
+.idea/
+src/gen/
+dist-ssr/
+
+*.log
+*.local
+.DS_Store

.gitignore

@@ -0,0 +1 @@
+gen/

Dockerfile

@@ -0,0 +1,6 @@
+FROM node:20-alpine
+WORKDIR /app
+
+EXPOSE 1234
+
+CMD ["sh", "-c", "npm install && node server.js"]

Dockerfile.prod

@@ -0,0 +1,29 @@
+FROM node:20-alpine AS builder
+
+RUN npm install -g grpc-tools protoc-gen-js
+
+WORKDIR /app
+COPY collab/package.json collab/package-lock.json* ./
+RUN npm install
+
+COPY proto ./proto
+
+RUN mkdir -p gen && \
+    PROTOC="/usr/local/bin/grpc_tools_node_protoc" && \
+    GRPC_PLUGIN="/usr/local/bin/grpc_tools_node_protoc_plugin" && \
+    "$PROTOC" -I proto \
+    --plugin=protoc-gen-grpc="$GRPC_PLUGIN" \
+    --grpc_out=grpc_js:gen \
+    --js_out=import_style=commonjs,binary:gen \
+    proto/yoresee_doc/v1/yoresee_doc.proto
+
+FROM node:20-alpine
+WORKDIR /app
+RUN mkdir -p src
+COPY --from=builder /app/node_modules ./node_modules
+COPY --from=builder /app/gen ./src/gen
+COPY collab/. .
+
+EXPOSE 1234
+
+CMD ["node", "server.js"]

package.json

@@ -0,0 +1,24 @@
+{
+  "name": "yoresee-doc-collab",
+  "version": "0.1.0",
+  "private": true,
+  "main": "server.js",
+  "type": "commonjs",
+  "scripts": {
+    "start": "node server.js"
+  },
+  "dependencies": {
+    "@grpc/grpc-js": "^1.12.0",
+    "@grpc/proto-loader": "^0.7.13",
+    "amqplib": "^0.10.9",
+    "google-protobuf": "^4.0.2",
+    "redis": "^4.6.0",
+    "ws": "^8.18.0",
+    "y-redis": "^1.0.3",
+    "y-websocket": "^1.5.4",
+    "yjs": "^13.6.16"
+  },
+  "devDependencies": {
+    "protoc": "^34.0.0"
+  }
+}

server.js

@@ -0,0 +1,36 @@
+const wsUtils = require('./src/ws-utils');
+const { setupWSConnection } = wsUtils;
+const config = require('./src/config');
+const redis = require('./src/redis');
+const grpc = require('./src/grpc');
+const mq = require('./src/mq');
+const { createHTTPServer } = require('./src/http/server');
+const { bindWebSocketGateway } = require('./src/ws-gateway');
+const { startServer } = require('./src/bootstrap');
+const { createGracefulShutdown, registerShutdownSignals } = require('./src/lifecycle');
+
+let isDraining = false;
+
+const server = createHTTPServer({
+  redis,
+  wsUtils,
+  getIsDraining: () => isDraining
+});
+const wss = bindWebSocketGateway({
+  server,
+  redis,
+  setupWSConnection
+});
+startServer({ server, config, redis, grpc });
+
+const shutdown = createGracefulShutdown({
+  server,
+  wss,
+  mq,
+  redis,
+  onDraining: () => {
+    isDraining = true;
+  }
+});
+
+registerShutdownSignals(shutdown);

src/bootstrap.js

@@ -0,0 +1,17 @@
+function startServer({ server, config, redis, grpc }) {
+  server.listen(config.port, async () => {
+    await redis.initRedis();
+    redis.initYRedis();
+    grpc.initGrpcClient();
+
+    await grpc.healthCheck();
+
+    console.log(`Collab server listening on ${config.port}`);
+    console.log(`Redis config: ${config.redisHost}:${config.redisPort}/${config.redisDb}`);
+    console.log('HTTP API: /api/active-rooms');
+  });
+}
+
+module.exports = {
+  startServer
+};

src/config.js

@@ -0,0 +1,31 @@
+const port = Number(process.env.PORT || 1234);
+
+const redisHost = process.env.REDIS_HOST || 'redis';
+const redisPort = Number(process.env.REDIS_PORT || 6379);
+const redisPassword = process.env.REDIS_PASSWORD || '';
+const redisDb = Number(process.env.REDIS_DB || 0);
+
+const backendAddr = process.env.BACKEND_ADDR || 'backend:9090';
+const dirtyDocTopic = process.env.DIRTY_DOC_TOPIC || 'collab.dirty_docs';
+const dirtyDocMqType = process.env.DIRTY_DOC_MQ || 'redis';
+const rabbitmqUrl = process.env.RABBITMQ_URL || '';
+const internalRpcKey = process.env.INTERNAL_RPC_KEY || '';
+const dirtyDocSetKey = process.env.DIRTY_DOC_SET_KEY || 'collab:yjs:dirty:doc';
+const dirtyDocNotifyThreshold = Number(process.env.DIRTY_DOC_NOTIFY_THRESHOLD || 200);
+const docUpdatesTTL = Number(process.env.DOC_UPDATES_TTL || 259200);
+
+module.exports = {
+  port,
+  redisHost,
+  redisPort,
+  redisPassword,
+  redisDb,
+  backendAddr,
+  dirtyDocTopic,
+  dirtyDocMqType,
+  rabbitmqUrl,
+  internalRpcKey,
+  dirtyDocSetKey,
+  dirtyDocNotifyThreshold,
+  docUpdatesTTL
+};

src/doc-loader.js

@@ -0,0 +1,96 @@
+const Y = require('yjs');
+const redis = require('./redis');
+const backend = require('./grpc');
+const { keyCollabDocUpdates } = require('./key');
+
+const snapshotTimeoutMs = Number(process.env.BACKEND_SNAPSHOT_TIMEOUT_MS || 3000);
+
+async function loadDoc(docId) {
+  const start = Date.now();
+  console.log(`[doc-loader] start docId=${docId}`);
+  const redisKey = keyCollabDocUpdates(docId);
+  const backendDocId = docId;
+
+  try {
+    const updates = await redis.getListBuffers(redisKey);
+    console.log(`[doc-loader] redis ${docId} ${updates && updates.length ? 'hit' : 'miss'}`);
+
+    const doc = new Y.Doc();
+    if (updates && updates.length > 0) {
+      for (const update of updates) {
+        if (update && update.length > 0) {
+          Y.applyUpdate(doc, update);
+        }
+      }
+      return doc;
+    }
+
+    console.log(`[doc-loader] no redis cache, trying snapshot docId=${docId}`);
+    try {
+      console.log(`[doc-loader] requesting snapshot docId=${docId}`);
+      const snapshot = await Promise.race([
+        backend.getDocumentYjsSnapshot(backendDocId),
+        new Promise((_, reject) => setTimeout(() => reject(new Error('snapshot timeout')), snapshotTimeoutMs))
+      ]);
+      console.log(`[doc-loader] received snapshot docId=${docId} length=${snapshot ? snapshot.length : 0}`);
+
+      if (snapshot && snapshot.length > 0) {
+        console.log(`[doc-loader] applying snapshot docId=${docId} bytes=${snapshot.length}`);
+        Y.applyUpdate(doc, snapshot);
+        console.log(`[doc-loader] saving snapshot to redis docId=${docId}`);
+        await redis.replaceListBuffer(redisKey, snapshot);
+        console.log(`[doc-loader] apply snapshot docId=${docId} bytes=${snapshot.length} took=${Date.now() - start}ms`);
+      } else {
+        console.log(`[doc-loader] snapshot empty docId=${docId} took=${Date.now() - start}ms`);
+        console.log(`[doc-loader] trying to get document content docId=${docId}`);
+        const content = await backend.getDocumentContent(backendDocId);
+        console.log(`[doc-loader] received content docId=${docId} length=${content ? content.length : 0}`);
+
+        if (content) {
+          console.log(`[doc-loader] seeding content docId=${docId} length=${content.length}`);
+          const ytext = doc.getText('content');
+          ytext.insert(0, content);
+          const initialUpdate = Y.encodeStateAsUpdate(doc);
+          console.log(`[doc-loader] saving initial update to redis docId=${docId}`);
+          await redis.replaceListBuffer(redisKey, initialUpdate);
+          console.log(`[doc-loader] seeded content docId=${docId} bytes=${initialUpdate.length} took=${Date.now() - start}ms`);
+        } else {
+          console.log(`[doc-loader] no content to seed docId=${docId} took=${Date.now() - start}ms`);
+        }
+      }
+    } catch (err) {
+      console.error(`[doc-loader] snapshot error docId=${docId} err=${err.message} stack=${err.stack} took=${Date.now() - start}ms`);
+      // Fallback to getting document content when snapshot fails
+      try {
+        console.log(`[doc-loader] fallback: trying to get document content docId=${docId}`);
+        const content = await backend.getDocumentContent(backendDocId);
+        console.log(`[doc-loader] fallback: received content docId=${docId} length=${content ? content.length : 0}`);
+
+        if (content) {
+          console.log(`[doc-loader] fallback: seeding content docId=${docId} length=${content.length}`);
+          const ytext = doc.getText('content');
+          ytext.insert(0, content);
+          const initialUpdate = Y.encodeStateAsUpdate(doc);
+          console.log(`[doc-loader] fallback: saving initial update to redis docId=${docId}`);
+          await redis.replaceListBuffer(redisKey, initialUpdate);
+          console.log(`[doc-loader] fallback seeded content docId=${docId} bytes=${initialUpdate.length} took=${Date.now() - start}ms`);
+        } else {
+          console.log(`[doc-loader] fallback no content to seed docId=${docId} took=${Date.now() - start}ms`);
+        }
+      } catch (fallbackErr) {
+        console.error(`[doc-loader] fallback error docId=${docId} err=${fallbackErr.message} stack=${fallbackErr.stack} took=${Date.now() - start}ms`);
+      }
+    }
+
+    console.log(`[doc-loader] returning doc docId=${docId} took=${Date.now() - start}ms`);
+    return doc;
+  } catch (err) {
+    console.error(`[doc-loader] fatal error docId=${docId} err=${err.message} stack=${err.stack} took=${Date.now() - start}ms`);
+    // Return empty doc as last resort
+    return new Y.Doc();
+  }
+}
+
+module.exports = {
+  loadDoc
+};

src/grpc/client.js

@@ -0,0 +1,37 @@
+const grpc = require('@grpc/grpc-js');
+const { DocumentServiceClient, SystemServiceClient } = require('../gen/yoresee_doc/v1/yoresee_doc_grpc_pb');
+const config = require('../config');
+
+let documentServiceClient = null;
+let systemServiceClient = null;
+
+const buildMetadata = () => {
+  const md = new grpc.Metadata();
+  if (config.internalRpcKey) {
+    md.set('x-internal-key', config.internalRpcKey);
+  }
+  return md;
+};
+
+function initGrpcClient() {
+  try {
+    const opts = {
+      'grpc.default_service_config': JSON.stringify({
+        loadBalancingConfig: [{ round_robin: {} }]
+      })
+    };
+    const creds = grpc.credentials.createInsecure();
+    documentServiceClient = new DocumentServiceClient(config.backendAddr, creds, opts);
+    systemServiceClient = new SystemServiceClient(config.backendAddr, creds, opts);
+    console.log(`gRPC clients initialized, connecting to ${config.backendAddr}`);
+  } catch (err) {
+    console.error('Failed to initialize gRPC client:', err.message);
+  }
+}
+
+module.exports = {
+  initGrpcClient,
+  buildMetadata,
+  getDocumentServiceClient: () => documentServiceClient,
+  getSystemServiceClient: () => systemServiceClient,
+};

src/grpc/document.js

@@ -0,0 +1,76 @@
+const pb = require('../gen/yoresee_doc/v1/yoresee_doc_pb');
+const { getDocumentServiceClient, buildMetadata } = require('./client');
+
+async function getDocumentContent(documentExternalId) {
+  return new Promise((resolve, reject) => {
+    const client = getDocumentServiceClient();
+    if (!client) {
+      return reject(new Error('gRPC client not initialized'));
+    }
+    const request = new pb.GetDocumentContentRequest();
+    request.setDocumentExternalId(documentExternalId);
+
+    client.getDocumentContent(request, buildMetadata(), (err, response) => {
+      if (err) {
+        return reject(err);
+      }
+      const base = response?.getBase?.();
+      if (base && base.getCode && base.getCode() !== 0) {
+        return reject(new Error(base.getMessage() || 'Failed to get document content'));
+      }
+      resolve(response?.getContent?.() || '');
+    });
+  });
+}
+
+async function getDocumentYjsSnapshot(documentExternalId) {
+  return new Promise((resolve, reject) => {
+    const client = getDocumentServiceClient();
+    if (!client) {
+      return reject(new Error('gRPC client not initialized'));
+    }
+    const request = new pb.GetDocumentYjsSnapshotRequest();
+    request.setDocumentExternalId(documentExternalId);
+
+    client.getDocumentYjsSnapshot(request, buildMetadata(), (err, response) => {
+      if (err) {
+        return reject(err);
+      }
+      const base = response?.getBase?.();
+      if (base && base.getCode && base.getCode() !== 0) {
+        return reject(new Error(base.getMessage() || 'Failed to get document snapshot'));
+      }
+      const state = response?.getState_asU8?.();
+      resolve(state && state.length ? Buffer.from(state) : null);
+    });
+  });
+}
+
+async function saveDocumentYjsSnapshot(documentExternalId, state) {
+  return new Promise((resolve, reject) => {
+    const client = getDocumentServiceClient();
+    if (!client) {
+      return reject(new Error('gRPC client not initialized'));
+    }
+    const request = new pb.SaveDocumentYjsSnapshotRequest();
+    request.setDocumentExternalId(documentExternalId);
+    request.setState(state);
+
+    client.saveDocumentYjsSnapshot(request, buildMetadata(), (err, response) => {
+      if (err) {
+        return reject(err);
+      }
+      const base = response?.getBase?.();
+      if (base && base.getCode && base.getCode() !== 0) {
+        return reject(new Error(base.getMessage() || 'Failed to save document snapshot'));
+      }
+      resolve(true);
+    });
+  });
+}
+
+module.exports = {
+  getDocumentContent,
+  getDocumentYjsSnapshot,
+  saveDocumentYjsSnapshot,
+};

src/grpc/index.js

@@ -0,0 +1,17 @@
+const { initGrpcClient, getDocumentServiceClient, getSystemServiceClient } = require('./client');
+const { getDocumentContent, getDocumentYjsSnapshot, saveDocumentYjsSnapshot } = require('./document');
+const { healthCheck } = require('./system');
+
+module.exports = {
+  initGrpcClient,
+  getDocumentContent,
+  getDocumentYjsSnapshot,
+  saveDocumentYjsSnapshot,
+  healthCheck,
+  get documentServiceClient() {
+    return getDocumentServiceClient();
+  },
+  get systemServiceClient() {
+    return getSystemServiceClient();
+  },
+};

src/grpc/system.js

@@ -0,0 +1,32 @@
+const pb = require('../gen/yoresee_doc/v1/yoresee_doc_pb');
+const { getSystemServiceClient, buildMetadata } = require('./client');
+
+async function healthCheck() {
+  const client = getSystemServiceClient();
+  if (!client) {
+    console.log('System service client not initialized, skipping health check');
+    return;
+  }
+
+  try {
+    const request = new pb.HealthRequest();
+    client.health(request, buildMetadata(), (err, response) => {
+      if (err) {
+        console.error('Health check failed:', err);
+      } else {
+        const base = response?.getBase?.();
+        if (base && base.getCode && base.getCode() === 0) {
+          console.log('Health check successful');
+        } else {
+          console.error('Health check failed:', base ? base.getMessage() : 'Unknown error');
+        }
+      }
+    });
+  } catch (err) {
+    console.error('Failed to execute health check:', err.message);
+  }
+}
+
+module.exports = {
+  healthCheck,
+};

src/http/docs.js

@@ -0,0 +1,63 @@
+const Y = require('yjs');
+const { keyCollabDocUpdates } = require('../key');
+
+function findDocInMemory(docs, docId) {
+  if (!docs) {
+    return null;
+  }
+  return docs.get(`doc-${docId}`) || docs.get(docId) || null;
+}
+
+async function buildDocFromRedis(redis, docId) {
+  const redisKey = keyCollabDocUpdates(docId);
+  const updates = await redis.getListBuffers(redisKey);
+  if (!updates || updates.length === 0) {
+    return null;
+  }
+
+  const doc = new Y.Doc();
+  for (const update of updates) {
+    if (update && update.length > 0) {
+      Y.applyUpdate(doc, update);
+    }
+  }
+  return doc;
+}
+
+async function loadDocForRead(docs, redis, docId) {
+  const inMemoryDoc = findDocInMemory(docs, docId);
+  if (inMemoryDoc) {
+    return inMemoryDoc;
+  }
+  return buildDocFromRedis(redis, docId);
+}
+
+const XML_FRAGMENT_DOC_TYPES = new Set(['yoresee_rich_text']);
+
+function encodeSnapshotResponse(doc, docType) {
+  const update = Y.encodeStateAsUpdate(doc);
+  let content = '';
+
+  if (XML_FRAGMENT_DOC_TYPES.has(docType)) {
+    const fragment = doc.getXmlFragment('content');
+    content = fragment ? fragment.toString() : '';
+  } else {
+    const ytext = doc.getText('content');
+    content = ytext ? ytext.toString() : '';
+  }
+
+  return {
+    state: Buffer.from(update).toString('base64'),
+    content
+  };
+}
+
+function encodeBinaryUpdate(doc) {
+  return Buffer.from(Y.encodeStateAsUpdate(doc));
+}
+
+module.exports = {
+  loadDocForRead,
+  encodeSnapshotResponse,
+  encodeBinaryUpdate
+};

src/http/helpers.js

@@ -0,0 +1,20 @@
+function writeJson(res, status, payload) {
+  res.writeHead(status, { 'Content-Type': 'application/json' });
+  res.end(JSON.stringify(payload));
+}
+
+function writeText(res, status, body) {
+  res.writeHead(status);
+  res.end(body);
+}
+
+function getRequestURL(req) {
+  const host = req.headers && req.headers.host ? req.headers.host : 'localhost';
+  return new URL(req.url || '/', `http://${host}`);
+}
+
+module.exports = {
+  writeJson,
+  writeText,
+  getRequestURL
+};

src/http/probes.js

@@ -0,0 +1,45 @@
+const { writeJson } = require('./helpers');
+
+function buildReadinessState(isDraining, redisClient) {
+  const ready = !isDraining && !!redisClient;
+  return {
+    ready,
+    redisState: redisClient ? 'connected' : 'disconnected',
+    detail: isDraining ? 'server is draining' : undefined
+  };
+}
+
+function handleProbeRequest(pathname, res, { isDraining, redisClient }) {
+  if (pathname === '/livez') {
+    writeJson(res, 200, { status: 'ok' });
+    return true;
+  }
+
+  const state = buildReadinessState(isDraining, redisClient);
+
+  if (pathname === '/readyz') {
+    writeJson(res, state.ready ? 200 : 503, {
+      status: state.ready ? 'ok' : 'not_ready',
+      redis: state.redisState,
+      detail: state.detail
+    });
+    return true;
+  }
+
+  if (pathname === '/health') {
+    writeJson(res, 200, {
+      status: 'ok',
+      redis: state.redisState,
+      persistence: 'custom',
+      ready: state.ready ? 'true' : 'false',
+      detail: state.detail
+    });
+    return true;
+  }
+
+  return false;
+}
+
+module.exports = {
+  handleProbeRequest
+};

src/http/router.js

@@ -0,0 +1,77 @@
+const { writeJson, writeText, getRequestURL } = require('./helpers');
+const { handleProbeRequest } = require('./probes');
+const { loadDocForRead, encodeSnapshotResponse, encodeBinaryUpdate } = require('./docs');
+
+function decodeDocID(pathname, prefix) {
+  try {
+    return decodeURIComponent(pathname.slice(prefix.length));
+  } catch (_err) {
+    return '';
+  }
+}
+
+function createRequestHandler({ redis, wsUtils, getIsDraining }) {
+  const docs = wsUtils.docs;
+
+  return async function handleRequest(req, res) {
+    try {
+      const url = getRequestURL(req);
+      const pathname = url.pathname;
+
+      if (handleProbeRequest(pathname, res, { isDraining: getIsDraining(), redisClient: redis.redisClient })) {
+        return;
+      }
+
+      if (pathname === '/api/active-rooms') {
+        const rooms = await redis.getActiveRooms();
+        writeJson(res, 200, { rooms, count: rooms.length });
+        return;
+      }
+
+      if (pathname.startsWith('/internal/yjs/doc-snapshot/')) {
+        const docId = decodeDocID(pathname, '/internal/yjs/doc-snapshot/');
+        if (!docId) {
+          writeText(res, 400, 'doc id required');
+          return;
+        }
+
+        const docType = url.searchParams.get('type') || '';
+        const doc = await loadDocForRead(docs, redis, docId);
+        if (!doc) {
+          writeText(res, 404, 'doc not loaded');
+          return;
+        }
+
+        writeJson(res, 200, encodeSnapshotResponse(doc, docType));
+        return;
+      }
+
+      if (pathname.startsWith('/internal/yjs/doc/')) {
+        const docId = decodeDocID(pathname, '/internal/yjs/doc/');
+        if (!docId) {
+          writeText(res, 400, 'doc id required');
+          return;
+        }
+
+        const doc = await loadDocForRead(docs, redis, docId);
+        if (!doc) {
+          writeText(res, 404, 'doc not loaded');
+          return;
+        }
+
+        res.writeHead(200, { 'Content-Type': 'application/octet-stream' });
+        res.end(encodeBinaryUpdate(doc));
+        return;
+      }
+
+      writeText(res, 404, '');
+    } catch (err) {
+      console.error('Request handling failed:', err);
+      writeText(res, 500, 'internal error');
+    }
+  };
+}
+
+module.exports = {
+  createRequestHandler
+};

src/http/server.js

@@ -0,0 +1,18 @@
+const http = require('http');
+const { createRequestHandler } = require('./router');
+
+function createHTTPServer({ redis, wsUtils, getIsDraining }) {
+  const requestHandler = createRequestHandler({
+    redis,
+    wsUtils,
+    getIsDraining
+  });
+
+  return http.createServer((req, res) => {
+    void requestHandler(req, res);
+  });
+}
+
+module.exports = {
+  createHTTPServer
+};

src/key.js

@@ -0,0 +1,32 @@
+function keyCollabDocUpdates(docId) {
+  return `collab:yjs:doc:updates:${docId}`;
+}
+
+function keyCollabRoom(docId) {
+  return `collab:room:doc-${docId}`;
+}
+
+function keyCollabRoomRaw(roomName) {
+  return `collab:room:${roomName}`;
+}
+
+function keyCollabRoomPattern() {
+  return 'collab:room:*';
+}
+
+function parseRoomName(redisKey) {
+  return redisKey.slice('collab:room:'.length);
+}
+
+function keyCollabYjs(roomName) {
+  return `collab:yjs:${roomName}`;
+}
+
+module.exports = {
+  keyCollabDocUpdates,
+  keyCollabRoom,
+  keyCollabRoomRaw,
+  keyCollabRoomPattern,
+  parseRoomName,
+  keyCollabYjs,
+};

src/lifecycle.js

@@ -0,0 +1,72 @@
+function closeServerGracefully(server) {
+  return new Promise((resolve) => {
+    server.close(() => resolve());
+  });
+}
+
+function closeWssGracefully(wss) {
+  return new Promise((resolve) => {
+    wss.close(() => resolve());
+  });
+}
+
+function createGracefulShutdown({
+  server,
+  wss,
+  mq,
+  redis,
+  onDraining,
+  timeoutMs = 10000
+}) {
+  let shuttingDown = false;
+
+  return async function shutdown(signal) {
+    if (shuttingDown) {
+      return;
+    }
+    shuttingDown = true;
+    onDraining();
+    console.log(`Received ${signal}, start graceful shutdown`);
+
+    const forceExitTimer = setTimeout(() => {
+      console.error('Graceful shutdown timeout, force exiting');
+      process.exit(1);
+    }, timeoutMs);
+    forceExitTimer.unref();
+
+    await Promise.all([
+      closeWssGracefully(wss),
+      closeServerGracefully(server)
+    ]);
+
+    try {
+      await mq.close();
+    } catch (err) {
+      console.error('Failed to close MQ:', err);
+    }
+
+    try {
+      await redis.closeRedis();
+    } catch (err) {
+      console.error('Failed to close Redis:', err);
+    }
+
+    clearTimeout(forceExitTimer);
+    process.exit(0);
+  };
+}
+
+function registerShutdownSignals(shutdown) {
+  process.on('SIGTERM', () => {
+    void shutdown('SIGTERM');
+  });
+
+  process.on('SIGINT', () => {
+    void shutdown('SIGINT');
+  });
+}
+
+module.exports = {
+  createGracefulShutdown,
+  registerShutdownSignals
+};

src/mq.js

@@ -0,0 +1,83 @@
+const amqp = require('amqplib');
+const config = require('./config');
+const redis = require('./redis');
+
+let rabbitConn = null;
+let rabbitChannel = null;
+
+const shouldPublishRedis = () => {
+  const mode = (config.dirtyDocMqType || 'redis').toLowerCase();
+  return mode === 'redis' || mode === 'both';
+};
+
+const shouldPublishRabbit = () => {
+  const mode = (config.dirtyDocMqType || 'redis').toLowerCase();
+  return mode === 'rabbitmq' || mode === 'rabbit' || mode === 'both';
+};
+
+const getRabbitChannel = async () => {
+  if (!config.rabbitmqUrl) {
+    return null;
+  }
+  if (rabbitChannel) {
+    return rabbitChannel;
+  }
+  rabbitConn = await amqp.connect(config.rabbitmqUrl);
+  rabbitChannel = await rabbitConn.createChannel();
+  return rabbitChannel;
+};
+
+const publishRabbit = async (topic, payload) => {
+  const channel = await getRabbitChannel();
+  if (!channel) {
+    return;
+  }
+  await channel.assertExchange(topic, 'topic', { durable: true });
+  channel.publish(topic, '', Buffer.from(payload), {
+    contentType: 'application/json'
+  });
+};
+
+const publishRedis = async (topic, payload) => {
+  if (!redis.redisClient) {
+    return;
+  }
+  await redis.redisClient.publish(topic, payload);
+};
+
+const publishDirtyDoc = async (docId) => {
+  if (!docId) {
+    return;
+  }
+  const payload = JSON.stringify({ doc_id: docId, ts: Date.now() });
+  try {
+    if (shouldPublishRedis()) {
+      await publishRedis(config.dirtyDocTopic, payload);
+    }
+  } catch (err) {
+    console.error(`Failed to publish dirty doc ${docId} to redis:`, err.message);
+  }
+  try {
+    if (shouldPublishRabbit()) {
+      await publishRabbit(config.dirtyDocTopic, payload);
+    }
+  } catch (err) {
+    console.error(`Failed to publish dirty doc ${docId} to rabbitmq:`, err.message);
+  }
+};
+
+const close = async () => {
+  if (rabbitChannel) {
+    await rabbitChannel.close();
+    rabbitChannel = null;
+  }
+  if (rabbitConn) {
+    await rabbitConn.close();
+    rabbitConn = null;
+  }
+};
+
+module.exports = {
+  publishDirtyDoc,
+  close
+};

src/persistence.js

@@ -0,0 +1,55 @@
+const Y = require('yjs');
+const redis = require('./redis');
+const mq = require('./mq');
+const config = require('./config');
+const { keyCollabDocUpdates } = require('./key');
+
+const compactThreshold = parseInt(process.env.DOC_UPDATE_COMPACT_THRESHOLD, 10) || 1000;
+const dirtyNotifyThreshold = config.dirtyDocNotifyThreshold;
+
+const updateCounts = new Map();
+const notifyCounts = new Map();
+const dirtyLogged = new Set();
+
+const persistUpdate = async (docId, doc, update) => {
+  if (!docId) {
+    return;
+  }
+  const redisKey = keyCollabDocUpdates(docId);
+  await redis.appendListBuffer(redisKey, update);
+  await redis.addDirtyDoc(docId);
+  await redis.updateRoomActiveTime(`doc-${docId}`);
+  if (!dirtyLogged.has(docId)) {
+    dirtyLogged.add(docId);
+    console.log(`[collab] dirty marked docId=${docId} redisKey=${redisKey}`);
+  }
+
+  const nextNotify = (notifyCounts.get(docId) || 0) + 1;
+  if (nextNotify >= dirtyNotifyThreshold) {
+    await mq.publishDirtyDoc(docId);
+    notifyCounts.set(docId, 0);
+  } else {
+    notifyCounts.set(docId, nextNotify);
+  }
+
+  const nextCount = (updateCounts.get(docId) || 0) + 1;
+  if (nextCount >= compactThreshold) {
+    const merged = Y.encodeStateAsUpdate(doc);
+    await redis.replaceListBuffer(redisKey, merged);
+    updateCounts.set(docId, 0);
+  } else {
+    updateCounts.set(docId, nextCount);
+  }
+};
+
+const attachPersistence = (doc, docId) => {
+  if (doc._persistenceBound) {
+    return;
+  }
+  doc._persistenceBound = true;
+  doc.on('update', (update) => {
+    void persistUpdate(docId, doc, update);
+  });
+};
+
+module.exports = { persistUpdate, attachPersistence };

src/redis/client.js

@@ -0,0 +1,40 @@
+const { createClient } = require('redis');
+const config = require('../config');
+
+let redisClient = null;
+
+async function initRedis() {
+  redisClient = createClient({
+    socket: {
+      host: config.redisHost,
+      port: config.redisPort
+    },
+    password: config.redisPassword || undefined,
+    database: config.redisDb
+  });
+
+  redisClient.on('error', (err) => console.error('Redis Client Error', err));
+
+  try {
+    await redisClient.connect();
+    console.log(`Connected to Redis at ${config.redisHost}:${config.redisPort}`);
+  } catch (err) {
+    console.error('Failed to connect to Redis:', err);
+  }
+}
+
+function initYRedis() {
+  console.log('Yjs Redis persistence initialized (custom loader)');
+}
+
+async function closeRedis() {
+  if (redisClient) {
+    await redisClient.quit();
+  }
+}
+
+function getClient() {
+  return redisClient;
+}
+
+module.exports = { initRedis, initYRedis, closeRedis, getClient };

src/redis/document.js

@@ -0,0 +1,87 @@
+const { commandOptions } = require('redis');
+const { getClient } = require('./client');
+const config = require('../config');
+
+async function getListBuffers(key) {
+  const client = getClient();
+  if (!client) return [];
+  try {
+    const items = await client.lRange(commandOptions({ returnBuffers: true }), key, 0, -1);
+    return items || [];
+  } catch (err) {
+    console.error(`Failed to lrange buffer for ${key}:`, err.message);
+    return [];
+  }
+}
+
+async function appendListBuffer(key, buffer) {
+  const client = getClient();
+  if (!client) return;
+  try {
+    const payload = Buffer.isBuffer(buffer) ? buffer : Buffer.from(buffer);
+    await client.rPush(key, payload);
+    if (config.docUpdatesTTL > 0) {
+      await client.expire(key, config.docUpdatesTTL);
+    }
+  } catch (err) {
+    console.error(`Failed to rpush buffer for ${key}:`, err.message);
+  }
+}
+
+async function replaceListBuffer(key, buffer) {
+  const client = getClient();
+  if (!client) return;
+  try {
+    const payload = Buffer.isBuffer(buffer) ? buffer : Buffer.from(buffer);
+    const pipeline = client.multi();
+    pipeline.del(key);
+    pipeline.rPush(key, payload);
+    if (config.docUpdatesTTL > 0) {
+      pipeline.expire(key, config.docUpdatesTTL);
+    }
+    await pipeline.exec();
+  } catch (err) {
+    console.error(`Failed to replace list buffer for ${key}:`, err.message);
+  }
+}
+
+async function addDirtyDoc(docId) {
+  const client = getClient();
+  if (!client) return;
+  try {
+    await client.sAdd(config.dirtyDocSetKey, docId);
+  } catch (err) {
+    console.error(`Failed to add dirty doc ${docId}:`, err.message);
+  }
+}
+
+async function removeDirtyDoc(docId) {
+  const client = getClient();
+  if (!client) return;
+  try {
+    await client.sRem(config.dirtyDocSetKey, docId);
+  } catch (err) {
+    console.error(`Failed to remove dirty doc ${docId}:`, err.message);
+  }
+}
+
+async function isDirtyDoc(docId) {
+  const client = getClient();
+  if (!client) return false;
+  try {
+    const exists = await client.sIsMember(config.dirtyDocSetKey, docId);
+    return !!exists;
+  } catch (err) {
+    console.error(`Failed to check dirty doc ${docId}:`, err.message);
+    return false;
+  }
+}
+
+module.exports = {
+  getListBuffers,
+  appendListBuffer,
+  replaceListBuffer,
+  addDirtyDoc,
+  removeDirtyDoc,
+  isDirtyDoc,
+};

src/redis/index.js

@@ -0,0 +1,21 @@
+const { initRedis, initYRedis, closeRedis, getClient } = require('./client');
+const { updateRoomActiveTime, getActiveRooms, checkRoomExists } = require('./room');
+const { getListBuffers, appendListBuffer, replaceListBuffer, addDirtyDoc, removeDirtyDoc, isDirtyDoc } = require('./document');
+
+module.exports = {
+  initRedis,
+  initYRedis,
+  closeRedis,
+  updateRoomActiveTime,
+  getActiveRooms,
+  checkRoomExists,
+  getListBuffers,
+  appendListBuffer,
+  replaceListBuffer,
+  addDirtyDoc,
+  removeDirtyDoc,
+  isDirtyDoc,
+  get redisClient() {
+    return getClient();
+  },
+};

src/redis/room.js

@@ -0,0 +1,43 @@
+const { getClient } = require('./client');
+const { keyCollabRoomRaw, keyCollabRoomPattern, parseRoomName, keyCollabYjs } = require('../key');
+
+async function updateRoomActiveTime(roomName) {
+  const client = getClient();
+  if (!client) return;
+
+  try {
+    const key = keyCollabRoomRaw(roomName);
+    await client.set(key, Date.now().toString(), { EX: 3600 * 24 });
+  } catch (err) {
+    console.error('Failed to update room active time:', err);
+  }
+}
+
+async function getActiveRooms() {
+  const client = getClient();
+  if (!client) return [];
+
+  try {
+    const keys = await client.keys(keyCollabRoomPattern());
+    return keys.map(k => parseRoomName(k));
+  } catch (err) {
+    console.error('Failed to get active rooms:', err);
+    return [];
+  }
+}
+
+async function checkRoomExists(roomName) {
+  const client = getClient();
+  if (!client) return false;
+
+  try {
+    const yjsKey = keyCollabYjs(roomName);
+    const exists = await client.exists(yjsKey);
+    return exists;
+  } catch (err) {
+    console.error(`Failed to check room ${roomName}:`, err.message);
+    return false;
+  }
+}
+
+module.exports = { updateRoomActiveTime, getActiveRooms, checkRoomExists };

src/utils.js

@@ -0,0 +1,38 @@
+const Y = require('yjs');
+const redis = require('./redis');
+const grpc = require('./grpc');
+
+async function checkAndInitRoom(roomName) {
+  if (!redis.redisClient || !grpc.documentServiceClient) {
+    return null;
+  }
+
+  try {
+    const exists = await redis.checkRoomExists(roomName);
+    if (exists) {
+      console.log(`Room ${roomName} already has data in Redis, skipping init`);
+      return null;
+    }
+
+    const docId = roomName.replace(/^doc-/, '');
+    console.log(`Room ${roomName} is empty, fetching content from backend for doc ${docId}...`);
+
+    const content = await grpc.getDocumentContent(docId);
+    if (content) {
+      console.log(`Got content from backend, length: ${content.length}`);
+      const ydoc = new Y.Doc();
+      const ytext = ydoc.getText('content');
+      ytext.insert(0, content);
+      return ydoc;
+    }
+
+    return null;
+  } catch (err) {
+    console.error(`Failed to check/init room ${roomName}:`, err.message);
+    return null;
+  }
+}
+
+module.exports = {
+  checkAndInitRoom
+};

src/ws-gateway.js

@@ -0,0 +1,29 @@
+const WebSocket = require('ws');
+
+function bindWebSocketGateway({ server, redis, setupWSConnection }) {
+  const wss = new WebSocket.Server({ server });
+
+  wss.on('connection', async (conn, req) => {
+    const url = new URL(req.url, `http://${req.headers.host}`);
+    const roomName = url.pathname.slice(1);
+
+    if (roomName) {
+      redis.updateRoomActiveTime(roomName);
+    }
+
+    try {
+      await setupWSConnection(conn, req, {
+        gc: true
+      });
+    } catch (err) {
+      console.error('Failed to setup WS connection:', err);
+      conn.close();
+    }
+  });
+
+  return wss;
+}
+
+module.exports = {
+  bindWebSocketGateway
+};

src/ws-utils.js

@@ -0,0 +1,222 @@
+const Y = require('yjs');
+const syncProtocol = require('y-protocols/dist/sync.cjs');
+const awarenessProtocol = require('y-protocols/dist/awareness.cjs');
+
+const encoding = require('lib0/dist/encoding.cjs');
+const decoding = require('lib0/dist/decoding.cjs');
+const { loadDoc } = require('./doc-loader');
+const { attachPersistence } = require('./persistence');
+
+const wsReadyStateConnecting = 0;
+const wsReadyStateOpen = 1;
+const wsReadyStateClosed = 3;
+
+const gcEnabled = process.env.GC !== 'false' && process.env.GC !== '0';
+
+const messageSync = 0;
+const messageAwareness = 1;
+
+const docs = new Map();
+const docPromises = new Map();
+
+exports.docs = docs;
+
+const extractDocId = (docName) => {
+  if (!docName) {
+    return '';
+  }
+  if (docName.startsWith('doc-')) {
+    return docName.slice(4);
+  }
+  return docName;
+};
+
+// ── WebSocket send / close ──────────────────────────────────
+
+const send = (doc, conn, m) => {
+  if (conn.readyState !== wsReadyStateConnecting && conn.readyState !== wsReadyStateOpen) {
+    closeConn(doc, conn);
+  }
+  try {
+    conn.send(m, err => {
+      if (err != null) {
+        closeConn(doc, conn);
+      }
+    });
+  } catch (err) {
+    closeConn(doc, conn);
+  }
+};
+
+const closeConn = (doc, conn) => {
+  if (doc.conns.has(conn)) {
+    const controlledIds = doc.conns.get(conn);
+    doc.conns.delete(conn);
+    awarenessProtocol.removeAwarenessStates(doc.awareness, Array.from(controlledIds), null);
+    if (doc.conns.size === 0) {
+      docs.delete(doc.name);
+      doc.destroy();
+    }
+  }
+  conn.close();
+};
+
+// ── WSSharedDoc ─────────────────────────────────────────────
+
+const updateHandler = (update, origin, doc) => {
+  const encoder = encoding.createEncoder();
+  encoding.writeVarUint(encoder, messageSync);
+  syncProtocol.writeUpdate(encoder, update);
+  const message = encoding.toUint8Array(encoder);
+  doc.conns.forEach((_, conn) => send(doc, conn, message));
+};
+
+class WSSharedDoc extends Y.Doc {
+  constructor(name) {
+    super({ gc: gcEnabled });
+    this.name = name;
+    this.conns = new Map();
+    this.awareness = new awarenessProtocol.Awareness(this);
+    this.awareness.setLocalState(null);
+
+    const awarenessChangeHandler = ({ added, updated, removed }, conn) => {
+      const changedClients = added.concat(updated, removed);
+      if (conn !== null) {
+        const connControlledIDs = this.conns.get(conn);
+        if (connControlledIDs !== undefined) {
+          added.forEach(clientID => { connControlledIDs.add(clientID); });
+          removed.forEach(clientID => { connControlledIDs.delete(clientID); });
+        }
+      }
+      const encoder = encoding.createEncoder();
+      encoding.writeVarUint(encoder, messageAwareness);
+      encoding.writeVarUint8Array(encoder, awarenessProtocol.encodeAwarenessUpdate(this.awareness, changedClients));
+      const buff = encoding.toUint8Array(encoder);
+      this.conns.forEach((_, c) => { send(this, c, buff); });
+    };
+    this.awareness.on('update', awarenessChangeHandler);
+    this.on('update', updateHandler);
+  }
+}
+
+// ── Document lifecycle ──────────────────────────────────────
+
+const createYDoc = async (docname, gc = true) => {
+  const doc = new WSSharedDoc(docname);
+  doc.gc = gc;
+
+  const docId = extractDocId(docname);
+  try {
+    const loadedDoc = await loadDoc(docId);
+    const update = Y.encodeStateAsUpdate(loadedDoc);
+    if (update && update.length > 0) {
+      Y.applyUpdate(doc, update);
+    }
+  } catch (err) {
+    console.error(`Failed to load doc ${docId}:`, err.message);
+  }
+
+  attachPersistence(doc, docId);
+  return doc;
+};
+
+const getYDoc = async (docname, gc = true) => {
+  if (docPromises.has(docname)) {
+    console.log(`[ws-utils] returning existing promise docname=${docname}`);
+    return docPromises.get(docname);
+  }
+  console.log(`[ws-utils] creating new doc docname=${docname}`);
+  const createPromise = createYDoc(docname, gc)
+    .finally(() => { docPromises.delete(docname); });
+  docPromises.set(docname, createPromise);
+  return createPromise;
+};
+
+// ── Message handler ─────────────────────────────────────────
+
+const messageListener = (conn, doc, message) => {
+  try {
+    const encoder = encoding.createEncoder();
+    const decoder = decoding.createDecoder(message);
+    const messageType = decoding.readVarUint(decoder);
+    switch (messageType) {
+      case messageSync:
+        encoding.writeVarUint(encoder, messageSync);
+        syncProtocol.readSyncMessage(decoder, encoder, doc, conn);
+        if (encoding.length(encoder) > 1) {
+          send(doc, conn, encoding.toUint8Array(encoder));
+        }
+        break;
+      case messageAwareness: {
+        awarenessProtocol.applyAwarenessUpdate(doc.awareness, decoding.readVarUint8Array(decoder), conn);
+        break;
+      }
+      default:
+        break;
+    }
+  } catch (err) {
+    console.error(err);
+    doc.emit('error', [err]);
+  }
+};
+
+// ── Connection setup ────────────────────────────────────────
+
+exports.setupWSConnection = async (conn, req, { docName = req.url.slice(1).split('?')[0], gc = true } = {}) => {
+  const pendingMessages = [];
+  let ready = false;
+  let closed = false;
+
+  let doc = null;
+
+  conn.on('message', message => {
+    if (!ready) {
+      pendingMessages.push(message);
+      return;
+    }
+    if (doc) {
+      messageListener(conn, doc, new Uint8Array(message));
+    }
+  });
+  conn.on('close', () => {
+    closed = true;
+    if (ready) {
+      closeConn(doc, conn);
+    }
+  });
+  conn.on('error', () => {
+    closed = true;
+    if (ready) {
+      closeConn(doc, conn);
+    }
+  });
+
+  doc = await getYDoc(docName, gc);
+  if (closed || conn.readyState === wsReadyStateClosed) {
+    return;
+  }
+
+  doc.conns.set(conn, new Set());
+  ready = true;
+
+  for (const message of pendingMessages) {
+    messageListener(conn, doc, new Uint8Array(message));
+  }
+  pendingMessages.length = 0;
+
+  const encoder = encoding.createEncoder();
+  encoding.writeVarUint(encoder, messageSync);
+  syncProtocol.writeSyncStep1(encoder, doc);
+  send(doc, conn, encoding.toUint8Array(encoder));
+
+  const awarenessStates = doc.awareness.getStates();
+  if (awarenessStates.size > 0) {
+    const awarenessEncoder = encoding.createEncoder();
+    encoding.writeVarUint(awarenessEncoder, messageAwareness);
+    encoding.writeVarUint8Array(
+      awarenessEncoder,
+      awarenessProtocol.encodeAwarenessUpdate(doc.awareness, Array.from(awarenessStates.keys()))
+    );
+    send(doc, conn, encoding.toUint8Array(awarenessEncoder));
+  }
+};