SpreadJS 协同服务器 MongoDB 数据库适配支持

为了支持 SpreadJS 协同编辑场景，协同服务器需要持久化存储文档、操作、快照及里程碑数据。本文介绍了 MongoDB 数据库适配器的实现方法，包括集合初始化、适配器接口实现以及里程碑存储支持。

一、MongoDB 集合初始化

协同编辑服务需要以下集合（Collections）来存储数据：

• documents：存储文档基本信息（类型、版本号、快照版本号）。
• operations：存储操作记录，用于实现基于 OT 的操作重放。
• snapshot_fragments：存储快照的分片数据，支持大文档分段管理。
• milestone_snapshot：存储里程碑快照，用于回溯和恢复。

初始化脚本示例如下：

export async function InitCollections(client) {const collectionsToCreate = [{ name: 'documents', indexes: [{ key: { id: 1 }, unique: true }] },{ name: 'operations', indexes: [{ key: { doc_id: 1, version: 1 }, unique: true }] },{ name: 'snapshot_fragments', indexes: [{ key: { doc_id: 1, fragment_id: 1 }, unique: true }] },{ name: 'milestone_snapshot', indexes: [{ key: { doc_id: 1, version: 1 }, unique: true }] },];await client.connect();const db = client.db(dbName);
const existingCollections = (await db.listCollections().toArray()).map(c => c.name);
for (const col of collectionsToCreate) {if (!existingCollections.includes(col.name)) {await db.createCollection(col.name);console.log(`Collection '${col.name}' created.`);} else {console.log(`Collection '${col.name}' already exists.`);}for (const idx of col.indexes) {await db.collection(col.name).createIndex(idx.key, { unique: idx.unique });}}
await client.close();
}

二、MongoDB 适配器实现

1. 适配器说明

适配器 MongoDb 继承了协同服务的数据库接口，负责：

• 文档信息存取
• 操作记录管理
• 快照存储与更新
• 分片快照管理

可根据业务需要，增加 事务（session） 或 并发冲突检测。

1. 适配器核心实现

export class MongoDb extends Db {constructor(client) {super();this.client = client;this.client.connect()this.db = client.db(dbName);}async getDocument(docId) {const documents = this.db.collection('documents');let row = await documents.findOne({ id: docId });if (row) {return {id: row.id,type: row.type,version: row.version,snapshotVersion: row.snapshot_version};}}async getSnapshot(docId) {const documents = this.db.collection('documents');let row = await documents.findOne({ id: docId });if (!row) {return null;}const fragments = await this.getFragments(docId);return {id: row.id,v: row.snapshot_version,type: row.type,fragments: fragments};}async getFragments(docId) {const fragments = this.db.collection('snapshot_fragments');const rows = await fragments.find({ doc_id: docId }).toArray();if (rows.length === 0) {return {};}const results = {};for (const row of rows) {results[row.fragment_id] = JSON.parse(row.data);}return results;}async getFragment(docId, fragmentId) {const fragments = this.db.collection('snapshot_fragments');const row = await fragments.findOne({ doc_id: docId, fragment_id: fragmentId });if (row) {return JSON.parse(row.data);}return null;}async getOps(docId, from, to) {const operations = this.db.collection('operations');const query = { doc_id: docId, version: { $gte: from } };if (to !== undefined) {query.version.$lte = to;}const rows = await operations.find(query).toArray();if (rows.length === 0) {return [];}return rows.map(row => JSON.parse(row.operation));}async commitOp(docId, op, document) {try {const documents = this.db.collection('documents');const operations = this.db.collection('operations');const row = await documents.findOne({ id: docId });if (op.create) {if (row) {throw new Error(`Document with id ${docId} already exists.`);}await documents.insertOne({id: docId,type: document.type,version: document.version,snapshot_version: document.snapshotVersion},);await operations.insertOne({doc_id: docId,version: op.v,operation: JSON.stringify(op)},);return true;}else if (op.del) {if (!row) {throw new Error(`Document with id ${docId} does not exist.`);}await documents.deleteOne({ id: docId },);return true;}else {if (!row || row.version !== op.v) {throw new Error(`Document with id ${docId} does not exist or version mismatch.`);}await operations.insertOne({doc_id: docId,version: op.v,operation: JSON.stringify(op)},);await documents.updateOne({ id: docId },{ $set: { version: document.version } },);return true;}}catch (error) {console.error('Error committing operation:', error);return false;}finally {}}async commitSnapshot(docId, snapshot) {try {const documents = this.db.collection('documents');const fragments = this.db.collection('snapshot_fragments');const row = await documents.findOne({ id: docId },);if (!row) {throw new Error(`Document with id ${docId} does not exist.`);}const currentSnapshotVersion = row.snapshot_version;if (snapshot.fromVersion !== currentSnapshotVersion || snapshot.v <= currentSnapshotVersion) {throw new Error(`Snapshot version mismatch: expected ${currentSnapshotVersion}, got ${snapshot.v}`);}await documents.updateOne({ id: docId },{ $set: { snapshot_version: snapshot.v } },);if (snapshot.fragmentsChanges.deleteSnapshot) {fragments.deleteMany({ doc_id: docId },);}else {const { createFragments, updateFragments, deleteFragments } = snapshot.fragmentsChanges;if (createFragments) {const createOps = Object.entries(createFragments).map(([id, data]) => ({doc_id: docId,fragment_id: id,data: JSON.stringify(data)}));if (createOps.length > 0) {await fragments.insertMany(createOps,);}}if (updateFragments) {const updateOps = Object.entries(updateFragments).map(([id, data]) => ({updateOne: {filter: { doc_id: docId, fragment_id: id },update: { $set: { data: JSON.stringify(data) } }}}));if (updateOps.length > 0) {await fragments.bulkWrite(updateOps,// { session });}}if (deleteFragments) {const deleteOps = deleteFragments.map(id => ({deleteOne: {filter: { doc_id: docId, fragment_id: id }}}));if (deleteOps.length > 0) {await fragments.bulkWrite(deleteOps,);}}}return true;}catch (error) {console.error('Error committing snapshot:', error);return false;}finally {}}async close() {await this.client.close();}
}

三、里程碑数据存储

里程碑快照用于优化快照恢复性能（避免从头重放所有操作）。 实现类 MongoMilestoneDb 提供 保存 与 读取 接口：

export class MongoMilestoneDb {constructor(client, interval) {this.client = client;this.interval = interval ? interval : 1000;this.db = client.db(dbName);}
async saveMilestoneSnapshot(snapshot) {const milestones = this.db.collection('milestone_snapshot');await milestones.insertOne({doc_id: snapshot.id,version: snapshot.v,snapshot: JSON.stringify(snapshot)});return true;}
async getMilestoneSnapshot(id, version) {const milestones = this.db.collection('milestone_snapshot');const row = await milestones.findOne({ doc_id: id, version: { $lte: version } },{ sort: { version: -1 } });if (row) {return JSON.parse(row.snapshot);}return null;}
}

四、在 DocumentServices 中配置

完成适配器与里程碑数据库后，需要在 DocumentServices 中进行配置：

const documentServices = new OT.DocumentServices(
{ db: new MongoDb(mongoClient),milestoneDb: new MongoMilestoneDb(mongoClient, 500)
});

这样，SpreadJS 协同服务器即可通过 MongoDB 实现文档存储、操作日志管理、快照与里程碑维护，保证协同编辑过程的高效与可扩展。

五、总结

本文展示了 SpreadJS 协同服务器对 MongoDB 数据库的适配实现，主要包括：

1. 集合初始化：定义所需集合与索引。
2. 数据库****适配器：支持文档、操作、快照的存储与管理。
3. 里程碑存储：提供快照的高效回溯能力。
4. 服务集成：在 DocumentServices 中配置 MongoDB 适配器。

借助 MongoDB 的高性能与灵活数据结构，SpreadJS 协同服务可实现稳定、可扩展的文档协作平台。

扩展链接

使用 MySQL 为 SpreadJS 协同服务器提供存储支持