SyncManager.js 31 KB

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