overleaf-cep/services/docstore/app/js/DocArchiveManager.js

import MongoManager from './MongoManager.js'
import Errors from './Errors.js'
import logger from '@overleaf/logger'
import Settings from '@overleaf/settings'
import crypto from 'node:crypto'
import { ReadableString } from '@overleaf/stream-utils'
import RangeManager from './RangeManager.js'
import PersistorManager from './PersistorManager.js'
import pMap from 'p-map'
import { streamToBuffer } from './StreamToBuffer.js'
import mongodb from 'mongodb-legacy'

const { BSON } = mongodb

const PARALLEL_JOBS = Settings.parallelArchiveJobs
const UN_ARCHIVE_BATCH_SIZE = Settings.unArchiveBatchSize

async function archiveAllDocs(projectId) {
  if (!_isArchivingEnabled()) {
    return
  }

  const docIds = await MongoManager.getNonArchivedProjectDocIds(projectId)
  await pMap(docIds, docId => archiveDoc(projectId, docId), {
    concurrency: PARALLEL_JOBS,
  })
}

async function archiveDoc(projectId, docId) {
  if (!_isArchivingEnabled()) {
    return
  }

  const doc = await MongoManager.getDocForArchiving(projectId, docId)

  if (!doc) {
    // The doc wasn't found, it was already archived, or the lock couldn't be
    // acquired. Since we don't know which it is, silently return.
    return
  }

  logger.debug({ projectId, docId: doc._id }, 'sending doc to persistor')
  const key = `${projectId}/${doc._id}`

  if (doc.lines == null) {
    throw new Error('doc has no lines')
  }

  RangeManager.fixCommentIds(doc)

  // warn about any oversized docs already in mongo
  const linesSize = BSON.calculateObjectSize(doc.lines || {})
  const rangesSize = BSON.calculateObjectSize(doc.ranges || {})
  if (
    linesSize > Settings.max_doc_length ||
    rangesSize > Settings.max_doc_length
  ) {
    logger.warn(
      { projectId, docId: doc._id, linesSize, rangesSize },
      'large doc found when archiving project'
    )
  }

  const json = JSON.stringify({
    lines: doc.lines,
    ranges: doc.ranges,
    rev: doc.rev,
    schema_v: 1,
  })

  // this should never happen, but protects against memory-corruption errors that
  // have happened in the past
  if (json.indexOf('\u0000') > -1) {
    const error = new Error('null bytes detected')
    logger.err({ err: error, doc }, error.message)
    throw error
  }

  const stream = new ReadableString(json)
  if (Settings.docstore.backend === 's3') {
    await PersistorManager.sendStream(Settings.docstore.bucket, key, stream)
  } else {
    await PersistorManager.sendStream(Settings.docstore.bucket, key, stream, {
      sourceMd5: crypto.createHash('md5').update(json).digest('hex'),
    })
  }

  await MongoManager.markDocAsArchived(projectId, docId, doc.rev)
}

async function unArchiveAllDocs(projectId) {
  if (!_isArchivingEnabled()) {
    return
  }

  while (true) {
    let docs
    if (Settings.docstore.keepSoftDeletedDocsArchived) {
      docs = await MongoManager.getNonDeletedArchivedProjectDocs(
        projectId,
        UN_ARCHIVE_BATCH_SIZE
      )
    } else {
      docs = await MongoManager.getArchivedProjectDocs(
        projectId,
        UN_ARCHIVE_BATCH_SIZE
      )
    }
    if (!docs || docs.length === 0) {
      break
    }
    await pMap(docs, doc => unarchiveDoc(projectId, doc._id), {
      concurrency: PARALLEL_JOBS,
    })
  }
}

// get the doc from the PersistorManager without storing it in mongo
async function getDoc(projectId, docId) {
  const key = `${projectId}/${docId}`
  const stream = await PersistorManager.getObjectStream(
    Settings.docstore.bucket,
    key
  )

  let buffer
  if (Settings.docstore.backend === 's3') {
    stream.resume()
    buffer = await streamToBuffer(projectId, docId, stream)
  } else {
    const sourceMd5 = await PersistorManager.getObjectMd5Hash(
      Settings.docstore.bucket,
      key
    )
    stream.resume()
    buffer = await streamToBuffer(projectId, docId, stream)
    const md5 = crypto.createHash('md5').update(buffer).digest('hex')
    if (sourceMd5 !== md5) {
      throw new Errors.Md5MismatchError('md5 mismatch when downloading doc', {
        key,
        sourceMd5,
        md5,
      })
    }
  }
  return _deserializeArchivedDoc(buffer)
}

// get the doc and unarchive it to mongo
async function unarchiveDoc(projectId, docId) {
  logger.debug({ projectId, docId }, 'getting doc from persistor')
  const mongoDoc = await MongoManager.findDoc(projectId, docId, {
    inS3: 1,
    rev: 1,
  })
  if (!mongoDoc.inS3) {
    // The doc is already unarchived
    return
  }

  if (!_isArchivingEnabled()) {
    throw new Error(
      'found archived doc, but archiving backend is not configured'
    )
  }

  const archivedDoc = await getDoc(projectId, docId)
  if (archivedDoc.rev == null) {
    // Older archived docs didn't have a rev. Assume that the rev of the
    // archived doc is the rev that was stored in Mongo when we retrieved it
    // earlier.
    archivedDoc.rev = mongoDoc.rev
  }
  await MongoManager.restoreArchivedDoc(projectId, docId, archivedDoc)
}

async function destroyProject(projectId) {
  const tasks = [MongoManager.destroyProject(projectId)]
  if (_isArchivingEnabled()) {
    tasks.push(
      PersistorManager.deleteDirectory(Settings.docstore.bucket, projectId)
    )
  }
  await Promise.all(tasks)
}

function _deserializeArchivedDoc(buffer) {
  const doc = JSON.parse(buffer)

  const result = {}
  if (doc.schema_v === 1 && doc.lines != null) {
    result.lines = doc.lines
    if (doc.ranges != null) {
      result.ranges = RangeManager.jsonRangesToMongo(doc.ranges)
    }
  } else if (Array.isArray(doc)) {
    result.lines = doc
  } else {
    throw new Error("I don't understand the doc format in s3")
  }

  if (doc.rev != null) {
    result.rev = doc.rev
  }

  return result
}

function _isArchivingEnabled() {
  const backend = Settings.docstore.backend

  if (!backend) {
    return false
  }

  // The default backend is S3. If another backend is configured or the S3
  // backend itself is correctly configured, then archiving is enabled.
  if (backend === 's3' && Settings.docstore.s3 == null) {
    return false
  }

  return true
}

export default {
  archiveAllDocs,
  archiveDoc,
  unArchiveAllDocs,
  unarchiveDoc,
  destroyProject,
  getDoc,
}