SyncManager.js 32 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097
  1. // @ts-check
  2. import _ from 'lodash'
  3. import { callbackify, promisify } from 'util'
  4. import { callbackifyMultiResult } from '@overleaf/promise-utils'
  5. import Settings from '@overleaf/settings'
  6. import logger from '@overleaf/logger'
  7. import Metrics from '@overleaf/metrics'
  8. import OError from '@overleaf/o-error'
  9. import { File, Range } from 'overleaf-editor-core'
  10. import { SyncError } from './Errors.js'
  11. import { db, ObjectId } from './mongodb.js'
  12. import * as SnapshotManager from './SnapshotManager.js'
  13. import * as LockManager from './LockManager.js'
  14. import * as UpdateTranslator from './UpdateTranslator.js'
  15. import * as UpdateCompressor from './UpdateCompressor.js'
  16. import * as WebApiManager from './WebApiManager.js'
  17. import * as ErrorRecorder from './ErrorRecorder.js'
  18. import * as RedisManager from './RedisManager.js'
  19. import * as HistoryStoreManager from './HistoryStoreManager.js'
  20. import * as HashManager from './HashManager.js'
  21. import { isInsert, isDelete } from './Utils.js'
  22. /**
  23. * @typedef {import('overleaf-editor-core').Comment} HistoryComment
  24. * @typedef {import('overleaf-editor-core').TrackedChange} HistoryTrackedChange
  25. * @typedef {import('./types').Comment} Comment
  26. * @typedef {import('./types').Entity} Entity
  27. * @typedef {import('./types').ResyncDocContentUpdate} ResyncDocContentUpdate
  28. * @typedef {import('./types').RetainOp} RetainOp
  29. * @typedef {import('./types').TrackedChange} TrackedChange
  30. * @typedef {import('./types').TrackedChangeTransition} TrackedChangeTransition
  31. * @typedef {import('./types').TrackingDirective} TrackingDirective
  32. * @typedef {import('./types').TrackingType} TrackingType
  33. * @typedef {import('./types').Update} Update
  34. */
  35. const MAX_RESYNC_HISTORY_RECORDS = 100 // keep this many records of previous resyncs
  36. const EXPIRE_RESYNC_HISTORY_INTERVAL_MS = 90 * 24 * 3600 * 1000 // 90 days
  37. const keys = Settings.redis.lock.key_schema
  38. // db.projectHistorySyncState.ensureIndex({expiresAt: 1}, {expireAfterSeconds: 0, background: true})
  39. // To add expiresAt field to existing entries in collection (choose a suitable future expiry date):
  40. // db.projectHistorySyncState.updateMany({resyncProjectStructure: false, resyncDocContents: [], expiresAt: {$exists:false}}, {$set: {expiresAt: new Date("2019-07-01")}})
  41. async function startResync(projectId, options = {}) {
  42. // We have three options here
  43. //
  44. // 1. If we update mongo before making the call to web then there's a
  45. // chance we ignore all updates indefinitely (there's no foolproff way
  46. // to undo the change in mongo)
  47. //
  48. // 2. If we make the call to web first then there is a small race condition
  49. // where we could process the sync update and then only update mongo
  50. // after, causing all updates to be ignored from then on
  51. //
  52. // 3. We can wrap everything in a project lock
  53. Metrics.inc('project_history_resync')
  54. try {
  55. await LockManager.promises.runWithLock(
  56. keys.projectHistoryLock({ project_id: projectId }),
  57. async extendLock => {
  58. await _startResyncWithoutLock(projectId, options)
  59. }
  60. )
  61. } catch (error) {
  62. // record error in starting sync ("sync ongoing")
  63. try {
  64. await ErrorRecorder.promises.record(projectId, -1, error)
  65. } catch (err) {
  66. // swallow any error thrown by ErrorRecorder.record()
  67. }
  68. throw error
  69. }
  70. }
  71. async function startHardResync(projectId, options = {}) {
  72. Metrics.inc('project_history_hard_resync')
  73. try {
  74. await LockManager.promises.runWithLock(
  75. keys.projectHistoryLock({ project_id: projectId }),
  76. async extendLock => {
  77. await clearResyncState(projectId)
  78. await RedisManager.promises.clearFirstOpTimestamp(projectId)
  79. await RedisManager.promises.destroyDocUpdatesQueue(projectId)
  80. await _startResyncWithoutLock(projectId, options)
  81. }
  82. )
  83. } catch (error) {
  84. // record error in starting sync ("sync ongoing")
  85. await ErrorRecorder.promises.record(projectId, -1, error)
  86. throw error
  87. }
  88. }
  89. async function _startResyncWithoutLock(projectId, options) {
  90. await ErrorRecorder.promises.recordSyncStart(projectId)
  91. const syncState = await _getResyncState(projectId)
  92. if (syncState.isSyncOngoing()) {
  93. throw new OError('sync ongoing')
  94. }
  95. syncState.setOrigin(options.origin || { kind: 'history-resync' })
  96. syncState.startProjectStructureSync()
  97. const webOpts = {}
  98. if (options.historyRangesMigration) {
  99. webOpts.historyRangesMigration = options.historyRangesMigration
  100. }
  101. await WebApiManager.promises.requestResync(projectId, webOpts)
  102. await setResyncState(projectId, syncState)
  103. }
  104. async function _getResyncState(projectId) {
  105. const rawSyncState = await db.projectHistorySyncState.findOne({
  106. project_id: new ObjectId(projectId.toString()),
  107. })
  108. const syncState = SyncState.fromRaw(projectId, rawSyncState)
  109. return syncState
  110. }
  111. async function setResyncState(projectId, syncState) {
  112. // skip if syncState is null (i.e. unchanged)
  113. if (syncState == null) {
  114. return
  115. }
  116. const update = {
  117. $set: syncState.toRaw(),
  118. $push: {
  119. history: {
  120. $each: [{ syncState: syncState.toRaw(), timestamp: new Date() }],
  121. $position: 0,
  122. $slice: MAX_RESYNC_HISTORY_RECORDS,
  123. },
  124. },
  125. $currentDate: { lastUpdated: true },
  126. }
  127. // handle different cases
  128. if (syncState.isSyncOngoing()) {
  129. // starting a new sync; prevent the entry expiring while sync is in ongoing
  130. update.$inc = { resyncCount: 1 }
  131. update.$unset = { expiresAt: true }
  132. } else {
  133. // successful completion of existing sync; set the entry to expire in the
  134. // future
  135. update.$set.expiresAt = new Date(
  136. Date.now() + EXPIRE_RESYNC_HISTORY_INTERVAL_MS
  137. )
  138. }
  139. // apply the update
  140. await db.projectHistorySyncState.updateOne(
  141. { project_id: new ObjectId(projectId) },
  142. update,
  143. { upsert: true }
  144. )
  145. }
  146. async function clearResyncState(projectId) {
  147. await db.projectHistorySyncState.deleteOne({
  148. project_id: new ObjectId(projectId.toString()),
  149. })
  150. }
  151. async function skipUpdatesDuringSync(projectId, updates) {
  152. const syncState = await _getResyncState(projectId)
  153. if (!syncState.isSyncOngoing()) {
  154. logger.debug({ projectId }, 'not skipping updates: no resync in progress')
  155. // don't return syncState when unchanged
  156. return { updates, syncState: null }
  157. }
  158. const filteredUpdates = []
  159. for (const update of updates) {
  160. syncState.updateState(update)
  161. const shouldSkipUpdate = syncState.shouldSkipUpdate(update)
  162. if (!shouldSkipUpdate) {
  163. filteredUpdates.push(update)
  164. } else {
  165. logger.debug({ projectId, update }, 'skipping update due to resync')
  166. }
  167. }
  168. return { updates: filteredUpdates, syncState }
  169. }
  170. async function expandSyncUpdates(
  171. projectId,
  172. projectHistoryId,
  173. updates,
  174. extendLock
  175. ) {
  176. const areSyncUpdatesQueued =
  177. _.some(updates, 'resyncProjectStructure') ||
  178. _.some(updates, 'resyncDocContent')
  179. if (!areSyncUpdatesQueued) {
  180. logger.debug({ projectId }, 'no resync updates to expand')
  181. return updates
  182. }
  183. const syncState = await _getResyncState(projectId)
  184. // compute the current snapshot from the most recent chunk
  185. const snapshotFiles = await SnapshotManager.promises.getLatestSnapshot(
  186. projectId,
  187. projectHistoryId
  188. )
  189. // check if snapshot files are valid
  190. const invalidFiles = _.pickBy(
  191. snapshotFiles,
  192. (v, k) => v == null || typeof v.isEditable !== 'function'
  193. )
  194. if (_.size(invalidFiles) > 0) {
  195. throw new SyncError('file is missing isEditable method', {
  196. projectId,
  197. invalidFiles,
  198. })
  199. }
  200. const expander = new SyncUpdateExpander(
  201. projectId,
  202. snapshotFiles,
  203. syncState.origin
  204. )
  205. // expand updates asynchronously to avoid blocking
  206. for (const update of updates) {
  207. await expander.expandUpdate(update)
  208. await extendLock()
  209. }
  210. return expander.getExpandedUpdates()
  211. }
  212. class SyncState {
  213. constructor(projectId, resyncProjectStructure, resyncDocContents, origin) {
  214. this.projectId = projectId
  215. this.resyncProjectStructure = resyncProjectStructure
  216. this.resyncDocContents = resyncDocContents
  217. this.origin = origin
  218. }
  219. static fromRaw(projectId, rawSyncState) {
  220. rawSyncState = rawSyncState || {}
  221. const resyncProjectStructure = rawSyncState.resyncProjectStructure || false
  222. const resyncDocContents = new Set(rawSyncState.resyncDocContents || [])
  223. const origin = rawSyncState.origin
  224. return new SyncState(
  225. projectId,
  226. resyncProjectStructure,
  227. resyncDocContents,
  228. origin
  229. )
  230. }
  231. toRaw() {
  232. return {
  233. resyncProjectStructure: this.resyncProjectStructure,
  234. resyncDocContents: Array.from(this.resyncDocContents),
  235. origin: this.origin,
  236. }
  237. }
  238. updateState(update) {
  239. if (update.resyncProjectStructure != null) {
  240. if (!this.isProjectStructureSyncing()) {
  241. throw new SyncError('unexpected resyncProjectStructure update', {
  242. projectId: this.projectId,
  243. resyncProjectStructure: this.resyncProjectStructure,
  244. })
  245. }
  246. if (this.isAnyDocContentSyncing()) {
  247. throw new SyncError('unexpected resyncDocContents update', {
  248. projectId: this.projectId,
  249. resyncDocContents: this.resyncDocContents,
  250. })
  251. }
  252. for (const doc of update.resyncProjectStructure.docs) {
  253. this.startDocContentSync(doc.path)
  254. }
  255. this.stopProjectStructureSync()
  256. } else if (update.resyncDocContent != null) {
  257. if (this.isProjectStructureSyncing()) {
  258. throw new SyncError('unexpected resyncDocContent update', {
  259. projectId: this.projectId,
  260. resyncProjectStructure: this.resyncProjectStructure,
  261. })
  262. }
  263. if (!this.isDocContentSyncing(update.path)) {
  264. throw new SyncError('unexpected resyncDocContent update', {
  265. projectId: this.projectId,
  266. resyncDocContents: this.resyncDocContents,
  267. path: update.path,
  268. })
  269. }
  270. this.stopDocContentSync(update.path)
  271. }
  272. }
  273. setOrigin(origin) {
  274. this.origin = origin
  275. }
  276. shouldSkipUpdate(update) {
  277. // don't skip sync updates
  278. if (
  279. update.resyncProjectStructure != null ||
  280. update.resyncDocContent != null
  281. ) {
  282. return false
  283. }
  284. // if syncing project structure skip update
  285. if (this.isProjectStructureSyncing()) {
  286. return true
  287. }
  288. // skip text updates for a docs being synced
  289. if (UpdateTranslator.isTextUpdate(update)) {
  290. if (this.isDocContentSyncing(update.meta.pathname)) {
  291. return true
  292. }
  293. }
  294. // preserve all other updates
  295. return false
  296. }
  297. startProjectStructureSync() {
  298. this.resyncProjectStructure = true
  299. this.resyncDocContents = new Set([])
  300. }
  301. stopProjectStructureSync() {
  302. this.resyncProjectStructure = false
  303. }
  304. stopDocContentSync(pathname) {
  305. this.resyncDocContents.delete(pathname)
  306. }
  307. startDocContentSync(pathname) {
  308. this.resyncDocContents.add(pathname)
  309. }
  310. isProjectStructureSyncing() {
  311. return this.resyncProjectStructure
  312. }
  313. isDocContentSyncing(pathname) {
  314. return this.resyncDocContents.has(pathname)
  315. }
  316. isAnyDocContentSyncing() {
  317. return this.resyncDocContents.size > 0
  318. }
  319. isSyncOngoing() {
  320. return this.isProjectStructureSyncing() || this.isAnyDocContentSyncing()
  321. }
  322. }
  323. class SyncUpdateExpander {
  324. /**
  325. * Build a SyncUpdateExpander
  326. *
  327. * @param {string} projectId
  328. * @param {Record<string, File>} snapshotFiles
  329. * @param {string} origin
  330. */
  331. constructor(projectId, snapshotFiles, origin) {
  332. this.projectId = projectId
  333. this.files = snapshotFiles
  334. this.expandedUpdates = []
  335. this.origin = origin
  336. }
  337. // If there's an expected *file* with the same path and either the same hash
  338. // or no hash, treat this as not editable even if history thinks it is.
  339. isEditable(filePath, file, expectedFiles) {
  340. if (!file.isEditable()) {
  341. return false
  342. }
  343. const fileHash = _.get(file, ['data', 'hash'])
  344. const matchedExpectedFile = expectedFiles.some(item => {
  345. const expectedFileHash = item._hash
  346. if (expectedFileHash && fileHash !== expectedFileHash) {
  347. // expected file has a hash and it doesn't match
  348. return false
  349. }
  350. return UpdateTranslator._convertPathname(item.path) === filePath
  351. })
  352. // consider editable file in history as binary, since it matches a binary file in the project
  353. return !matchedExpectedFile
  354. }
  355. /**
  356. * @param {Update} update
  357. */
  358. async expandUpdate(update) {
  359. if ('resyncProjectStructure' in update) {
  360. logger.debug(
  361. { projectId: this.projectId, update },
  362. 'expanding resyncProjectStructure update'
  363. )
  364. const persistedNonBinaryFileEntries = _.pickBy(this.files, (v, k) =>
  365. this.isEditable(k, v, update.resyncProjectStructure.files)
  366. )
  367. const persistedNonBinaryFiles = _.map(
  368. Object.keys(persistedNonBinaryFileEntries),
  369. path => ({
  370. path,
  371. })
  372. )
  373. const persistedBinaryFileEntries = _.omitBy(this.files, (v, k) =>
  374. this.isEditable(k, v, update.resyncProjectStructure.files)
  375. )
  376. // preserve file properties on binary files, for future comparison.
  377. const persistedBinaryFiles = _.map(
  378. persistedBinaryFileEntries,
  379. (entity, key) => Object.assign({}, entity, { path: key })
  380. )
  381. const expectedNonBinaryFiles = _.map(
  382. update.resyncProjectStructure.docs,
  383. entity =>
  384. Object.assign({}, entity, {
  385. path: UpdateTranslator._convertPathname(entity.path),
  386. })
  387. )
  388. const expectedBinaryFiles = _.map(
  389. update.resyncProjectStructure.files,
  390. entity =>
  391. Object.assign({}, entity, {
  392. path: UpdateTranslator._convertPathname(entity.path),
  393. })
  394. )
  395. // We need to detect and fix consistency issues where web and
  396. // history-store disagree on whether an entity is binary or not. Thus we
  397. // need to remove and add the two separately.
  398. this.queueRemoveOpsForUnexpectedFiles(
  399. update,
  400. expectedBinaryFiles,
  401. persistedBinaryFiles
  402. )
  403. this.queueRemoveOpsForUnexpectedFiles(
  404. update,
  405. expectedNonBinaryFiles,
  406. persistedNonBinaryFiles
  407. )
  408. this.queueAddOpsForMissingFiles(
  409. update,
  410. expectedBinaryFiles,
  411. persistedBinaryFiles
  412. )
  413. this.queueAddOpsForMissingFiles(
  414. update,
  415. expectedNonBinaryFiles,
  416. persistedNonBinaryFiles
  417. )
  418. this.queueUpdateForOutOfSyncBinaryFiles(
  419. update,
  420. expectedBinaryFiles,
  421. persistedBinaryFiles
  422. )
  423. } else if ('resyncDocContent' in update) {
  424. logger.debug(
  425. { projectId: this.projectId, update },
  426. 'expanding resyncDocContent update'
  427. )
  428. await this.expandResyncDocContentUpdate(update)
  429. } else {
  430. this.expandedUpdates.push(update)
  431. }
  432. }
  433. getExpandedUpdates() {
  434. return this.expandedUpdates
  435. }
  436. /**
  437. * @param {Entity[]} expectedFiles
  438. * @param {{ path: string }[]} persistedFiles
  439. */
  440. queueRemoveOpsForUnexpectedFiles(update, expectedFiles, persistedFiles) {
  441. const unexpectedFiles = _.differenceBy(
  442. persistedFiles,
  443. expectedFiles,
  444. 'path'
  445. )
  446. for (const entity of unexpectedFiles) {
  447. update = {
  448. pathname: entity.path,
  449. new_pathname: '',
  450. meta: {
  451. resync: true,
  452. origin: this.origin,
  453. ts: update.meta.ts,
  454. },
  455. }
  456. this.expandedUpdates.push(update)
  457. Metrics.inc('project_history_resync_operation', 1, {
  458. status: 'remove unexpected file',
  459. })
  460. }
  461. }
  462. /**
  463. * @param {Entity[]} expectedFiles
  464. * @param {{ path: string }[]} persistedFiles
  465. */
  466. queueAddOpsForMissingFiles(update, expectedFiles, persistedFiles) {
  467. const missingFiles = _.differenceBy(expectedFiles, persistedFiles, 'path')
  468. for (const entity of missingFiles) {
  469. update = {
  470. pathname: entity.path,
  471. meta: {
  472. resync: true,
  473. origin: this.origin,
  474. ts: update.meta.ts,
  475. },
  476. }
  477. if ('doc' in entity) {
  478. update.doc = entity.doc
  479. update.docLines = ''
  480. // we have to create a dummy entry here because later we will need the content in the diff computation
  481. this.files[update.pathname] = File.fromString('')
  482. } else {
  483. update.file = entity.file
  484. update.url = entity.url
  485. }
  486. this.expandedUpdates.push(update)
  487. Metrics.inc('project_history_resync_operation', 1, {
  488. status: 'add missing file',
  489. })
  490. }
  491. }
  492. queueUpdateForOutOfSyncBinaryFiles(update, expectedFiles, persistedFiles) {
  493. // create a map to lookup persisted files by their path
  494. const persistedFileMap = new Map(persistedFiles.map(x => [x.path, x]))
  495. // now search for files with same path but different hash values
  496. const differentFiles = expectedFiles.filter(expected => {
  497. // check for a persisted file at the same path
  498. const expectedPath = expected.path
  499. const persistedFileAtSamePath = persistedFileMap.get(expectedPath)
  500. if (!persistedFileAtSamePath) return false
  501. // check if the persisted file at the same path has a different hash
  502. const expectedHash = _.get(expected, '_hash')
  503. const persistedHash = _.get(persistedFileAtSamePath, ['data', 'hash'])
  504. const hashesPresent = expectedHash && persistedHash
  505. return hashesPresent && persistedHash !== expectedHash
  506. })
  507. for (const entity of differentFiles) {
  508. // remove the outdated persisted file
  509. const removeUpdate = {
  510. pathname: entity.path,
  511. new_pathname: '',
  512. meta: {
  513. resync: true,
  514. origin: this.origin,
  515. ts: update.meta.ts,
  516. },
  517. }
  518. this.expandedUpdates.push(removeUpdate)
  519. // add the new file content
  520. const addUpdate = {
  521. pathname: entity.path,
  522. meta: {
  523. resync: true,
  524. origin: this.origin,
  525. ts: update.meta.ts,
  526. },
  527. file: entity.file,
  528. url: entity.url,
  529. }
  530. this.expandedUpdates.push(addUpdate)
  531. Metrics.inc('project_history_resync_operation', 1, {
  532. status: 'update binary file contents',
  533. })
  534. }
  535. }
  536. /**
  537. * Expand a resyncDocContentUpdate
  538. *
  539. * @param {ResyncDocContentUpdate} update
  540. */
  541. async expandResyncDocContentUpdate(update) {
  542. const pathname = UpdateTranslator._convertPathname(update.path)
  543. const snapshotFile = this.files[pathname]
  544. const expectedFile = update.resyncDocContent
  545. const expectedContent = expectedFile.content
  546. if (!snapshotFile) {
  547. throw new OError('unrecognised file: not in snapshot')
  548. }
  549. // Compare hashes to see if the persisted file matches the expected content.
  550. // The hash of the persisted files is stored in the snapshot.
  551. // Note getHash() returns the hash only when the persisted file has
  552. // no changes in the snapshot, the hash is null if there are changes
  553. // that apply to it.
  554. let hashesMatch = false
  555. const persistedHash = snapshotFile.getHash()
  556. if (persistedHash != null) {
  557. const expectedHash = HashManager._getBlobHashFromString(expectedContent)
  558. if (persistedHash === expectedHash) {
  559. logger.debug(
  560. { projectId: this.projectId, persistedHash, expectedHash },
  561. 'skipping diff because hashes match and persisted file has no ops'
  562. )
  563. hashesMatch = true
  564. }
  565. } else {
  566. logger.debug('cannot compare hashes, will retrieve content')
  567. }
  568. // compute the difference between the expected and persisted content
  569. const historyId = await WebApiManager.promises.getHistoryId(this.projectId)
  570. const file = await snapshotFile.load(
  571. 'eager',
  572. HistoryStoreManager.getBlobStore(historyId)
  573. )
  574. const persistedContent = file.getContent()
  575. if (persistedContent == null) {
  576. // This should not happen given that we loaded the file eagerly. We could
  577. // probably refine the types in overleaf-editor-core so that this check
  578. // wouldn't be necessary.
  579. throw new Error('File was not properly loaded')
  580. }
  581. if (!hashesMatch) {
  582. const expandedUpdate = await this.queueUpdateForOutOfSyncContent(
  583. update,
  584. pathname,
  585. persistedContent,
  586. expectedContent
  587. )
  588. if (expandedUpdate != null) {
  589. // Adjust the ranges for the changes that have been made to the content
  590. for (const op of expandedUpdate.op) {
  591. if (isInsert(op)) {
  592. file.getComments().applyInsert(new Range(op.p, op.i.length))
  593. file.getTrackedChanges().applyInsert(op.p, op.i)
  594. } else if (isDelete(op)) {
  595. file.getComments().applyDelete(new Range(op.p, op.d.length))
  596. file.getTrackedChanges().applyDelete(op.p, op.d.length)
  597. }
  598. }
  599. }
  600. }
  601. const persistedComments = file.getComments().toArray()
  602. await this.queueUpdatesForOutOfSyncComments(
  603. update,
  604. pathname,
  605. persistedComments
  606. )
  607. const persistedChanges = file.getTrackedChanges().asSorted()
  608. await this.queueUpdatesForOutOfSyncTrackedChanges(
  609. update,
  610. pathname,
  611. persistedChanges
  612. )
  613. }
  614. /**
  615. * Queue update for out of sync content
  616. *
  617. * @param {ResyncDocContentUpdate} update
  618. * @param {string} pathname
  619. * @param {string} persistedContent
  620. * @param {string} expectedContent
  621. */
  622. async queueUpdateForOutOfSyncContent(
  623. update,
  624. pathname,
  625. persistedContent,
  626. expectedContent
  627. ) {
  628. logger.debug(
  629. { projectId: this.projectId, persistedContent, expectedContent },
  630. 'diffing doc contents'
  631. )
  632. const op = UpdateCompressor.diffAsShareJsOps(
  633. persistedContent,
  634. expectedContent
  635. )
  636. if (op.length === 0) {
  637. return null
  638. }
  639. const expandedUpdate = {
  640. doc: update.doc,
  641. op,
  642. meta: {
  643. resync: true,
  644. origin: this.origin,
  645. ts: update.meta.ts,
  646. pathname,
  647. doc_length: persistedContent.length,
  648. },
  649. }
  650. logger.debug(
  651. { projectId: this.projectId, diffCount: op.length },
  652. 'doc contents differ'
  653. )
  654. this.expandedUpdates.push(expandedUpdate)
  655. Metrics.inc('project_history_resync_operation', 1, {
  656. status: 'update text file contents',
  657. })
  658. return expandedUpdate
  659. }
  660. /**
  661. * Queue updates for out of sync comments
  662. *
  663. * @param {ResyncDocContentUpdate} update
  664. * @param {string} pathname
  665. * @param {HistoryComment[]} persistedComments
  666. */
  667. async queueUpdatesForOutOfSyncComments(update, pathname, persistedComments) {
  668. const expectedContent = update.resyncDocContent.content
  669. const expectedComments = update.resyncDocContent.ranges?.comments ?? []
  670. const resolvedCommentIds = new Set(
  671. update.resyncDocContent.resolvedCommentIds ?? []
  672. )
  673. const expectedCommentsById = new Map(
  674. expectedComments.map(comment => [comment.id, comment])
  675. )
  676. const persistedCommentsById = new Map(
  677. persistedComments.map(comment => [comment.id, comment])
  678. )
  679. // Delete any persisted comment that is not in the expected comment list.
  680. for (const persistedComment of persistedComments) {
  681. if (!expectedCommentsById.has(persistedComment.id)) {
  682. this.expandedUpdates.push({
  683. pathname,
  684. deleteComment: persistedComment.id,
  685. meta: {
  686. resync: true,
  687. origin: this.origin,
  688. ts: update.meta.ts,
  689. },
  690. })
  691. }
  692. }
  693. for (const expectedComment of expectedComments) {
  694. const persistedComment = persistedCommentsById.get(expectedComment.id)
  695. const expectedCommentResolved = resolvedCommentIds.has(expectedComment.id)
  696. if (
  697. persistedComment != null &&
  698. commentRangesAreInSync(persistedComment, expectedComment)
  699. ) {
  700. if (expectedCommentResolved === persistedComment.resolved) {
  701. // Both comments are identical; do nothing
  702. } else {
  703. // Only the resolved state differs
  704. this.expandedUpdates.push({
  705. pathname,
  706. commentId: expectedComment.id,
  707. resolved: expectedCommentResolved,
  708. meta: {
  709. resync: true,
  710. origin: this.origin,
  711. ts: update.meta.ts,
  712. },
  713. })
  714. }
  715. } else {
  716. const op = { ...expectedComment.op, resolved: expectedCommentResolved }
  717. // New comment or ranges differ
  718. this.expandedUpdates.push({
  719. doc: update.doc,
  720. op: [op],
  721. meta: {
  722. resync: true,
  723. origin: this.origin,
  724. ts: update.meta.ts,
  725. pathname,
  726. doc_length: expectedContent.length,
  727. },
  728. })
  729. }
  730. }
  731. }
  732. /**
  733. * Queue updates for out of sync tracked changes
  734. *
  735. * @param {ResyncDocContentUpdate} update
  736. * @param {string} pathname
  737. * @param {readonly HistoryTrackedChange[]} persistedChanges
  738. */
  739. async queueUpdatesForOutOfSyncTrackedChanges(
  740. update,
  741. pathname,
  742. persistedChanges
  743. ) {
  744. const expectedChanges = update.resyncDocContent.ranges?.changes ?? []
  745. const expectedContent = update.resyncDocContent.content
  746. /**
  747. * A cursor on the expected content
  748. */
  749. let cursor = 0
  750. /**
  751. * The persisted tracking at cursor
  752. *
  753. * @type {TrackingDirective}
  754. */
  755. let persistedTracking = { type: 'none' }
  756. /**
  757. * The expected tracking at cursor
  758. *
  759. * @type {TrackingDirective}
  760. */
  761. let expectedTracking = { type: 'none' }
  762. /**
  763. * The retain ops for the update
  764. *
  765. * @type {RetainOp[]}
  766. */
  767. const ops = []
  768. /**
  769. * The retain op being built
  770. *
  771. * @type {RetainOp | null}
  772. */
  773. let currentOp = null
  774. for (const transition of getTrackedChangesTransitions(
  775. persistedChanges,
  776. expectedChanges,
  777. expectedContent.length
  778. )) {
  779. if (transition.pos > cursor) {
  780. // The next transition will move the cursor. Decide what to do with the interval.
  781. if (trackingDirectivesEqual(expectedTracking, persistedTracking)) {
  782. // Expected tracking and persisted tracking are in sync. Emit the
  783. // current op and skip this interval.
  784. if (currentOp != null) {
  785. ops.push(currentOp)
  786. currentOp = null
  787. }
  788. } else {
  789. // Expected tracking and persisted tracking are different.
  790. const retainedText = expectedContent.slice(cursor, transition.pos)
  791. if (
  792. currentOp?.tracking != null &&
  793. trackingDirectivesEqual(expectedTracking, currentOp.tracking)
  794. ) {
  795. // The current op has the right tracking. Extend it.
  796. currentOp.r += retainedText
  797. } else {
  798. // The current op doesn't have the right tracking. Emit the current
  799. // op and start a new one.
  800. if (currentOp != null) {
  801. ops.push(currentOp)
  802. }
  803. currentOp = {
  804. r: retainedText,
  805. p: cursor,
  806. tracking: expectedTracking,
  807. }
  808. }
  809. }
  810. // Advance cursor
  811. cursor = transition.pos
  812. }
  813. // Update the expected and persisted tracking
  814. if (transition.stage === 'persisted') {
  815. persistedTracking = transition.tracking
  816. } else {
  817. expectedTracking = transition.tracking
  818. }
  819. }
  820. // Emit the last op
  821. if (currentOp != null) {
  822. ops.push(currentOp)
  823. }
  824. if (ops.length > 0) {
  825. this.expandedUpdates.push({
  826. doc: update.doc,
  827. op: ops,
  828. meta: {
  829. resync: true,
  830. origin: this.origin,
  831. ts: update.meta.ts,
  832. pathname,
  833. doc_length: expectedContent.length,
  834. },
  835. })
  836. }
  837. }
  838. }
  839. /**
  840. * Compares the ranges in the persisted and expected comments
  841. *
  842. * @param {HistoryComment} persistedComment
  843. * @param {Comment} expectedComment
  844. */
  845. function commentRangesAreInSync(persistedComment, expectedComment) {
  846. const expectedPos = expectedComment.op.hpos ?? expectedComment.op.p
  847. const expectedLength = expectedComment.op.hlen ?? expectedComment.op.c.length
  848. if (expectedLength === 0) {
  849. // A zero length comment from RangesManager is a detached comment in history
  850. return persistedComment.ranges.length === 0
  851. }
  852. if (persistedComment.ranges.length !== 1) {
  853. // The editor only supports single range comments
  854. return false
  855. }
  856. const persistedRange = persistedComment.ranges[0]
  857. return (
  858. persistedRange.pos === expectedPos &&
  859. persistedRange.length === expectedLength
  860. )
  861. }
  862. /**
  863. * Iterates through expected tracked changes and persisted tracked changes and
  864. * returns all transitions, sorted by position.
  865. *
  866. * @param {readonly HistoryTrackedChange[]} persistedChanges
  867. * @param {TrackedChange[]} expectedChanges
  868. * @param {number} docLength
  869. */
  870. function getTrackedChangesTransitions(
  871. persistedChanges,
  872. expectedChanges,
  873. docLength
  874. ) {
  875. /** @type {TrackedChangeTransition[]} */
  876. const transitions = []
  877. for (const change of persistedChanges) {
  878. transitions.push({
  879. stage: 'persisted',
  880. pos: change.range.start,
  881. tracking: {
  882. type: change.tracking.type,
  883. userId: change.tracking.userId,
  884. ts: change.tracking.ts.toISOString(),
  885. },
  886. })
  887. transitions.push({
  888. stage: 'persisted',
  889. pos: change.range.end,
  890. tracking: { type: 'none' },
  891. })
  892. }
  893. for (const change of expectedChanges) {
  894. const op = change.op
  895. const pos = op.hpos ?? op.p
  896. if (isInsert(op)) {
  897. transitions.push({
  898. stage: 'expected',
  899. pos,
  900. tracking: {
  901. type: 'insert',
  902. userId: change.metadata.user_id,
  903. ts: change.metadata.ts,
  904. },
  905. })
  906. transitions.push({
  907. stage: 'expected',
  908. pos: pos + op.i.length,
  909. tracking: { type: 'none' },
  910. })
  911. } else {
  912. transitions.push({
  913. stage: 'expected',
  914. pos,
  915. tracking: {
  916. type: 'delete',
  917. userId: change.metadata.user_id,
  918. ts: change.metadata.ts,
  919. },
  920. })
  921. transitions.push({
  922. stage: 'expected',
  923. pos: pos + op.d.length,
  924. tracking: { type: 'none' },
  925. })
  926. }
  927. }
  928. transitions.push({
  929. stage: 'expected',
  930. pos: docLength,
  931. tracking: { type: 'none' },
  932. })
  933. transitions.sort((a, b) => {
  934. if (a.pos < b.pos) {
  935. return -1
  936. } else if (a.pos > b.pos) {
  937. return 1
  938. } else if (a.tracking.type === 'none' && b.tracking.type !== 'none') {
  939. // none type comes before other types so that it can be overridden at the
  940. // same position
  941. return -1
  942. } else if (a.tracking.type !== 'none' && b.tracking.type === 'none') {
  943. // none type comes before other types so that it can be overridden at the
  944. // same position
  945. return 1
  946. } else {
  947. return 0
  948. }
  949. })
  950. return transitions
  951. }
  952. /**
  953. * Returns true if both tracking directives are equal
  954. *
  955. * @param {TrackingDirective} a
  956. * @param {TrackingDirective} b
  957. */
  958. function trackingDirectivesEqual(a, b) {
  959. if (a.type === 'none') {
  960. return b.type === 'none'
  961. } else {
  962. return a.type === b.type && a.userId === b.userId && a.ts === b.ts
  963. }
  964. }
  965. // EXPORTS
  966. const startResyncCb = callbackify(startResync)
  967. const startHardResyncCb = callbackify(startHardResync)
  968. const setResyncStateCb = callbackify(setResyncState)
  969. const clearResyncStateCb = callbackify(clearResyncState)
  970. const skipUpdatesDuringSyncCb = callbackifyMultiResult(skipUpdatesDuringSync, [
  971. 'updates',
  972. 'syncState',
  973. ])
  974. const expandSyncUpdatesCb = (
  975. projectId,
  976. projectHistoryId,
  977. updates,
  978. extendLock,
  979. callback
  980. ) => {
  981. const extendLockPromises = promisify(extendLock)
  982. expandSyncUpdates(projectId, projectHistoryId, updates, extendLockPromises)
  983. .then(result => {
  984. callback(null, result)
  985. })
  986. .catch(err => {
  987. callback(err)
  988. })
  989. }
  990. export {
  991. startResyncCb as startResync,
  992. startHardResyncCb as startHardResync,
  993. setResyncStateCb as setResyncState,
  994. clearResyncStateCb as clearResyncState,
  995. skipUpdatesDuringSyncCb as skipUpdatesDuringSync,
  996. expandSyncUpdatesCb as expandSyncUpdates,
  997. }
  998. export const promises = {
  999. startResync,
  1000. startHardResync,
  1001. setResyncState,
  1002. clearResyncState,
  1003. skipUpdatesDuringSync,
  1004. expandSyncUpdates,
  1005. }