Overleaf/services/history-v1/storage/lib/chunk_store/postgres.js

const { Chunk } = require('overleaf-editor-core')
const assert = require('../assert')
const knex = require('../knex')
const { ChunkVersionConflictError } = require('./errors')

const DUPLICATE_KEY_ERROR_CODE = '23505'

/**
 * Get the latest chunk's metadata from the database
 */
async function getLatestChunk(projectId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  const record = await knex('chunks')
    .where('doc_id', projectId)
    .orderBy('end_version', 'desc')
    .first()
  if (record == null) {
    return null
  }
  return chunkFromRecord(record)
}

/**
 * Get the metadata for the chunk that contains the given version.
 */
async function getChunkForVersion(projectId, version) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  const record = await knex('chunks')
    .where('doc_id', projectId)
    .where('end_version', '>=', version)
    .orderBy('end_version')
    .first()
  if (!record) {
    throw new Chunk.VersionNotFoundError(projectId, version)
  }
  return chunkFromRecord(record)
}

/**
 * Get the metadata for the chunk that contains the version that was current at
 * the given timestamp.
 */
async function getChunkForTimestamp(projectId, timestamp) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  // This query will find the latest chunk after the timestamp (query orders
  // in reverse chronological order), OR the latest chunk
  // This accounts for the case where the timestamp is ahead of the chunk's
  // timestamp and therefore will not return any results
  const whereAfterEndTimestampOrLatestChunk = knex.raw(
    'end_timestamp >= ? ' +
      'OR id = ( ' +
      'SELECT id FROM chunks ' +
      'WHERE doc_id = ? ' +
      'ORDER BY end_version desc LIMIT 1' +
      ')',
    [timestamp, projectId]
  )

  const record = await knex('chunks')
    .where('doc_id', projectId)
    .where(whereAfterEndTimestampOrLatestChunk)
    .orderBy('end_version')
    .first()
  if (!record) {
    throw new Chunk.BeforeTimestampNotFoundError(projectId, timestamp)
  }
  return chunkFromRecord(record)
}

/**
 * Build a chunk metadata object from the database record
 */
function chunkFromRecord(record) {
  return {
    id: record.id,
    startVersion: record.start_version,
    endVersion: record.end_version,
  }
}

/**
 * Get all of a project's chunk ids
 */
async function getProjectChunkIds(projectId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  const records = await knex('chunks').select('id').where('doc_id', projectId)
  return records.map(record => record.id)
}

/**
 * Insert a pending chunk before sending it to object storage.
 */
async function insertPendingChunk(projectId, chunk) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  const result = await knex.first(
    knex.raw("nextval('chunks_id_seq'::regclass)::integer as chunkid")
  )
  const chunkId = result.chunkid
  await knex('pending_chunks').insert({
    id: chunkId,
    doc_id: projectId,
    end_version: chunk.getEndVersion(),
    start_version: chunk.getStartVersion(),
    end_timestamp: chunk.getEndTimestamp(),
  })
  return chunkId
}

/**
 * Record that a new chunk was created.
 */
async function confirmCreate(projectId, chunk, chunkId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  await knex.transaction(async tx => {
    await Promise.all([
      _deletePendingChunk(tx, projectId, chunkId),
      _insertChunk(tx, projectId, chunk, chunkId),
    ])
  })
}

/**
 * Record that a chunk was replaced by a new one.
 */
async function confirmUpdate(projectId, oldChunkId, newChunk, newChunkId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  await knex.transaction(async tx => {
    await _deleteChunks(tx, { doc_id: projectId, id: oldChunkId })
    await Promise.all([
      _deletePendingChunk(tx, projectId, newChunkId),
      _insertChunk(tx, projectId, newChunk, newChunkId),
    ])
  })
}

async function _deletePendingChunk(tx, projectId, chunkId) {
  await tx('pending_chunks')
    .where({
      doc_id: projectId,
      id: chunkId,
    })
    .del()
}

async function _insertChunk(tx, projectId, chunk, chunkId) {
  const startVersion = chunk.getStartVersion()
  const endVersion = chunk.getEndVersion()
  try {
    await tx('chunks').insert({
      id: chunkId,
      doc_id: projectId,
      start_version: startVersion,
      end_version: endVersion,
      end_timestamp: chunk.getEndTimestamp(),
    })
  } catch (err) {
    if (err.code === DUPLICATE_KEY_ERROR_CODE) {
      throw new ChunkVersionConflictError(
        'chunk start or end version is not unique',
        { projectId, chunkId, startVersion, endVersion }
      )
    }
    throw err
  }
}

/**
 * Delete a chunk.
 *
 * @param {number} projectId
 * @param {number} chunkId
 * @return {Promise}
 */
async function deleteChunk(projectId, chunkId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')
  assert.integer(chunkId, 'bad chunkId')

  await _deleteChunks(knex, { doc_id: projectId, id: chunkId })
}

/**
 * Delete all of a project's chunks
 */
async function deleteProjectChunks(projectId) {
  projectId = parseInt(projectId, 10)
  assert.integer(projectId, 'bad projectId')

  knex.transaction(async tx => {
    await _deleteChunks(knex, { doc_id: projectId })
  })
}

async function _deleteChunks(tx, whereClause) {
  const rows = await tx('chunks').returning('*').where(whereClause).del()
  if (rows.length === 0) {
    return
  }

  const oldChunks = rows.map(row => ({
    doc_id: row.doc_id,
    chunk_id: row.id,
    start_version: row.start_version,
    end_version: row.end_version,
    end_timestamp: row.end_timestamp,
    deleted_at: tx.fn.now(),
  }))
  await tx('old_chunks').insert(oldChunks)
}

/**
 * Get a batch of old chunks for deletion
 */
async function getOldChunksBatch(count, minAgeSecs) {
  const maxDeletedAt = new Date(Date.now() - minAgeSecs * 1000)
  const records = await knex('old_chunks')
    .whereNull('deleted_at')
    .orWhere('deleted_at', '<', maxDeletedAt)
    .orderBy('chunk_id')
    .limit(count)
  return records.map(oldChunk => ({
    projectId: oldChunk.doc_id.toString(),
    chunkId: oldChunk.chunk_id,
  }))
}

/**
 * Delete a batch of old chunks from the database
 */
async function deleteOldChunks(chunkIds) {
  await knex('old_chunks').whereIn('chunk_id', chunkIds).del()
}

/**
 * Generate a new project id
 */
async function generateProjectId() {
  const record = await knex.first(
    knex.raw("nextval('docs_id_seq'::regclass)::integer as doc_id")
  )
  return record.doc_id.toString()
}

module.exports = {
  getLatestChunk,
  getChunkForVersion,
  getChunkForTimestamp,
  getProjectChunkIds,
  insertPendingChunk,
  confirmCreate,
  confirmUpdate,
  deleteChunk,
  deleteProjectChunks,
  getOldChunksBatch,
  deleteOldChunks,
  generateProjectId,
}