<commit_before>/* eslint-env browser */

// Module to upgrade the schema of messages, e.g. migrate attachments to disk.
// `processAll` purposely doesn’t rely on our Backbone IndexedDB adapter to
// prevent automatic migrations. Rather, it uses direct IndexedDB access.
// This includes avoiding usage of `storage` module which uses Backbone under
// the hood.

const isFunction = require('lodash/isFunction');
const isNumber = require('lodash/isNumber');
const isObject = require('lodash/isObject');
const isString = require('lodash/isString');
const last = require('lodash/last');

const Message = require('./types/message');
const { deferredToPromise } = require('./deferred_to_promise');


const MESSAGES_STORE_NAME = 'messages';
const ITEMS_STORE_NAME = 'items';
const NUM_MESSAGES_PER_BATCH = 50;

exports.processNext = async ({
  BackboneMessage,
  BackboneMessageCollection,
  count,
  upgradeMessageSchema,
} = {}) => {
  if (!isFunction(BackboneMessage)) {
    throw new TypeError('"BackboneMessage" (Whisper.Message) constructor is required');
  }

  if (!isFunction(BackboneMessageCollection)) {
    throw new TypeError('"BackboneMessageCollection" (Whisper.MessageCollection)' +
      ' constructor is required');
  }

  if (!isNumber(count)) {
    throw new TypeError('"count" is required');
  }

  if (!isFunction(upgradeMessageSchema)) {
    throw new TypeError('"upgradeMessageSchema" is required');
  }

  const startTime = Date.now();

  const fetchStartTime = Date.now();
  const messagesRequiringSchemaUpgrade =
    await _fetchMessagesRequiringSchemaUpgrade({ BackboneMessageCollection, count });
  const fetchDuration = Date.now() - fetchStartTime;

  const upgradeStartTime = Date.now();
  const upgradedMessages =
    await Promise.all(messagesRequiringSchemaUpgrade.map(upgradeMessageSchema));
  const upgradeDuration = Date.now() - upgradeStartTime;

  const saveStartTime = Date.now();
  const saveMessage = _saveMessageBackbone({ BackboneMessage });
  await Promise.all(upgradedMessages.map(saveMessage));
  const saveDuration = Date.now() - saveStartTime;

  const totalDuration = Date.now() - startTime;
  const numProcessed = messagesRequiringSchemaUpgrade.length;
  const hasMore = numProcessed > 0;
  return {
    hasMore,
    numProcessed,
    fetchDuration,
    upgradeDuration,
    saveDuration,
    totalDuration,
  };
};

exports.processAll = async ({
    Backbone,
    databaseName,
    databaseVersion,
    upgradeMessageSchema,
  } = {}) => {
  if (!isObject(Backbone)) {
    throw new TypeError('"Backbone" is required');
  }

  if (!isString(databaseName)) {
    throw new TypeError('"databaseName" must be a string');
  }

  if (!isNumber(databaseVersion)) {
    throw new TypeError('"databaseVersion" must be a number');
  }

  if (!isFunction(upgradeMessageSchema)) {
    throw new TypeError('"upgradeMessageSchema" is required');
  }

  const connection = await openDatabase(databaseName, databaseVersion);
  const isComplete = await isMigrationComplete(connection);
  console.log('Is attachment migration complete?', isComplete);
  if (isComplete) {
    return;
  }

  const migrationStartTime = Date.now();
  let unprocessedMessages = [];
  do {
    // eslint-disable-next-line no-await-in-loop
    const lastProcessedIndex = await getLastProcessedIndex(connection);

    const fetchUnprocessedMessagesStartTime = Date.now();
    unprocessedMessages =
      // eslint-disable-next-line no-await-in-loop
      await _dangerouslyFetchMessagesRequiringSchemaUpgradeWithoutIndex({
        connection,
        count: NUM_MESSAGES_PER_BATCH,
        lastIndex: lastProcessedIndex,
      });
    const fetchDuration = Date.now() - fetchUnprocessedMessagesStartTime;
    const numUnprocessedMessages = unprocessedMessages.length;

    if (numUnprocessedMessages === 0) {
      break;
    }

    const upgradeStartTime = Date.now();
    const upgradedMessages =
      // eslint-disable-next-line no-await-in-loop
      await Promise.all(unprocessedMessages.map(upgradeMessageSchema));
    const upgradeDuration = Date.now() - upgradeStartTime;

    const saveMessagesStartTime = Date.now();
    const transaction = connection.transaction(MESSAGES_STORE_NAME, 'readwrite');
    const transactionCompletion = completeTransaction(transaction);
    // eslint-disable-next-line no-await-in-loop
    await Promise.all(upgradedMessages.map(_saveMessage({ transaction })));
    // eslint-disable-next-line no-await-in-loop
    await transactionCompletion;
    const saveDuration = Date.now() - saveMessagesStartTime;

    // TODO: Confirm transaction is complete

    const lastMessage = last(upgradedMessages);
    const newLastProcessedIndex = lastMessage ? lastMessage.id : null;
    if (newLastProcessedIndex) {
      // eslint-disable-next-line no-await-in-loop
      await setLastProcessedIndex(connection, newLastProcessedIndex);
    }

    console.log('Upgrade message schema on startup:', {
      lastProcessedIndex,
      numUnprocessedMessages,
      fetchDuration,
      saveDuration,
      upgradeDuration,
      newLastProcessedIndex,
    });
  } while (unprocessedMessages.length > 0);

  await markMigrationComplete(connection);
  connection.close();

  const totalDuration = Date.now() - migrationStartTime;
  console.log('Attachment migration complete:', { totalDuration });
};

const _saveMessageBackbone = ({ BackboneMessage } = {}) => (message) => {
  const backboneMessage = new BackboneMessage(message);
  return deferredToPromise(backboneMessage.save());
};

const _saveMessage = ({ transaction } = {}) => (message) => {
  if (!isObject(transaction)) {
    throw new TypeError('"transaction" is required');
  }

  const messagesStore = transaction.objectStore(MESSAGES_STORE_NAME);
  const request = messagesStore.put(message, message.id);<commit_msg>Log total number of processed messages<commit_after>/* eslint-env browser */

// Module to upgrade the schema of messages, e.g. migrate attachments to disk.
// `processAll` purposely doesn’t rely on our Backbone IndexedDB adapter to
// prevent automatic migrations. Rather, it uses direct IndexedDB access.
// This includes avoiding usage of `storage` module which uses Backbone under
// the hood.

const isFunction = require('lodash/isFunction');
const isNumber = require('lodash/isNumber');
const isObject = require('lodash/isObject');
const isString = require('lodash/isString');
const last = require('lodash/last');

const Message = require('./types/message');
const { deferredToPromise } = require('./deferred_to_promise');


const MESSAGES_STORE_NAME = 'messages';
const ITEMS_STORE_NAME = 'items';
const NUM_MESSAGES_PER_BATCH = 50;

exports.processNext = async ({
  BackboneMessage,
  BackboneMessageCollection,
  count,
  upgradeMessageSchema,
} = {}) => {
  if (!isFunction(BackboneMessage)) {
    throw new TypeError('"BackboneMessage" (Whisper.Message) constructor is required');
  }

  if (!isFunction(BackboneMessageCollection)) {
    throw new TypeError('"BackboneMessageCollection" (Whisper.MessageCollection)' +
      ' constructor is required');
  }

  if (!isNumber(count)) {
    throw new TypeError('"count" is required');
  }

  if (!isFunction(upgradeMessageSchema)) {
    throw new TypeError('"upgradeMessageSchema" is required');
  }

  const startTime = Date.now();

  const fetchStartTime = Date.now();
  const messagesRequiringSchemaUpgrade =
    await _fetchMessagesRequiringSchemaUpgrade({ BackboneMessageCollection, count });
  const fetchDuration = Date.now() - fetchStartTime;

  const upgradeStartTime = Date.now();
  const upgradedMessages =
    await Promise.all(messagesRequiringSchemaUpgrade.map(upgradeMessageSchema));
  const upgradeDuration = Date.now() - upgradeStartTime;

  const saveStartTime = Date.now();
  const saveMessage = _saveMessageBackbone({ BackboneMessage });
  await Promise.all(upgradedMessages.map(saveMessage));
  const saveDuration = Date.now() - saveStartTime;

  const totalDuration = Date.now() - startTime;
  const numProcessed = messagesRequiringSchemaUpgrade.length;
  const hasMore = numProcessed > 0;
  return {
    hasMore,
    numProcessed,
    fetchDuration,
    upgradeDuration,
    saveDuration,
    totalDuration,
  };
};

exports.processAll = async ({
    Backbone,
    databaseName,
    databaseVersion,
    upgradeMessageSchema,
  } = {}) => {
  if (!isObject(Backbone)) {
    throw new TypeError('"Backbone" is required');
  }

  if (!isString(databaseName)) {
    throw new TypeError('"databaseName" must be a string');
  }

  if (!isNumber(databaseVersion)) {
    throw new TypeError('"databaseVersion" must be a number');
  }

  if (!isFunction(upgradeMessageSchema)) {
    throw new TypeError('"upgradeMessageSchema" is required');
  }

  const connection = await openDatabase(databaseName, databaseVersion);
  const isComplete = await isMigrationComplete(connection);
  console.log('Is attachment migration complete?', isComplete);
  if (isComplete) {
    return;
  }

  const migrationStartTime = Date.now();
  let unprocessedMessages = [];
  let totalMessagesProcessed = 0;
  do {
    // eslint-disable-next-line no-await-in-loop
    const lastProcessedIndex = await getLastProcessedIndex(connection);

    const fetchUnprocessedMessagesStartTime = Date.now();
    unprocessedMessages =
      // eslint-disable-next-line no-await-in-loop
      await _dangerouslyFetchMessagesRequiringSchemaUpgradeWithoutIndex({
        connection,
        count: NUM_MESSAGES_PER_BATCH,
        lastIndex: lastProcessedIndex,
      });
    const fetchDuration = Date.now() - fetchUnprocessedMessagesStartTime;
    const numUnprocessedMessages = unprocessedMessages.length;

    if (numUnprocessedMessages === 0) {
      break;
    }

    const upgradeStartTime = Date.now();
    const upgradedMessages =
      // eslint-disable-next-line no-await-in-loop
      await Promise.all(unprocessedMessages.map(upgradeMessageSchema));
    const upgradeDuration = Date.now() - upgradeStartTime;

    const saveMessagesStartTime = Date.now();
    const transaction = connection.transaction(MESSAGES_STORE_NAME, 'readwrite');
    const transactionCompletion = completeTransaction(transaction);
    // eslint-disable-next-line no-await-in-loop
    await Promise.all(upgradedMessages.map(_saveMessage({ transaction })));
    // eslint-disable-next-line no-await-in-loop
    await transactionCompletion;
    const saveDuration = Date.now() - saveMessagesStartTime;

    // TODO: Confirm transaction is complete

    const lastMessage = last(upgradedMessages);
    const newLastProcessedIndex = lastMessage ? lastMessage.id : null;
    if (newLastProcessedIndex) {
      // eslint-disable-next-line no-await-in-loop
      await setLastProcessedIndex(connection, newLastProcessedIndex);
    }

    totalMessagesProcessed += numUnprocessedMessages;
    console.log('Upgrade message schema:', {
      lastProcessedIndex,
      numUnprocessedMessages,
      numCumulativeMessagesProcessed: totalMessagesProcessed,
      fetchDuration,
      saveDuration,
      upgradeDuration,
      newLastProcessedIndex,
    });
  } while (unprocessedMessages.length > 0);

  await markMigrationComplete(connection);
  connection.close();

  const totalDuration = Date.now() - migrationStartTime;
  console.log('Attachment migration complete:', {
    totalDuration,
    totalMessagesProcessed,
  });
};

const _saveMessageBackbone = ({ BackboneMessage } = {}) => (message) => {
  const backboneMessage = new BackboneMessage(message);
  return deferredToPromise(backboneMessage.save());
};

const _saveMessage = ({ transaction } = {}) => (message) => {
  if (!isObject(transaction)) {
    throw new TypeError('"transaction" is required');
  }

  const messagesStore = transaction.objectStore(MESSAGES_STORE_NAME);
  const request = messagesStore.put(message, message.id);
{'fraction_non_alphanumeric': 0.10960643250105798, 'fraction_numerical': 0.0009310198899703766, 'mean_word_length': 4.3081761006289305, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2870943', 'n_tokens_mistral': 3371, 'n_tokens_neox': 3292, 'n_words': 756}