<commit_before>import { dependencies as Inject, singleton as Singleton } from "needlepoint";
import Db from "mongodb";
import ScheduledTask from "../ScheduledTask";
import Configuration from "../../core/config/Configuration";
import ChatApi from "../../core/api/ChatApi";
import Threads from "../../core/config/Threads";
import Promise from "bluebird";
import until from "async/until";

@Singleton
@Inject(Configuration, Db, ChatApi, Threads)
export default class ThreadHistoryDownloader extends ScheduledTask {
    /**
     * @param {Configuration} config
     * @param {Db} db
     * @param {ChatApi} api
     * @param {Threads} threads
     */
    constructor(config, db, api, threads) {
        super();

        this.config = config;
        this.db = db;
        this.api = api;
        this.threads = threads;
    }

    execute() {
        const threadIds = this.threads.getThreadIds();

        const promises = threadIds.map(id => this.downloadMessagesByThreadId(id));

        return Promise.all(promises);
    }

    downloadMessagesByThreadId(threadId) {
        const dbCollection = this.db.collection(`thread-${threadId}`);

        //region promises
        //region messageID index
        const msgIdIndexName = "messageID";
        const uniqueMsgIdIndex = {};
        uniqueMsgIdIndex[msgIdIndexName] = 1;

        const createIndexPromise = dbCollection.indexExists(msgIdIndexName)
            .then(() => Promise.resolve())
            .catch((err) => dbCollection.createIndex(uniqueMsgIdIndex, { unique: true }));
        //endregion

        const dbMsgCountPromise = dbCollection.count();
        const fbMsgCountPromise = this.api.getThreadInfo(threadId).then(info => info.messageCount);
        //endregion

        return Promise
            .all([createIndexPromise, dbMsgCountPromise, fbMsgCountPromise])
            .spread((createIndex, dbCount, fbCount) => {
                console.log(`Thread ${threadId}: There are ${dbCount} messages in cache, and ${fbCount} reported by Facebook`);

                if (dbCount == fbCount) {
                    return Promise.reject(new ThreadHistoryDownloadCancelledError(
                        `Thread ${threadId}: Message cache is up-to-date`
                    ));
                }

                if (dbCount > fbCount) {
                    console.warn(`There are more messages in the cache (${dbCount}) than Facebook reports (${fbCount}). ` +
                        `Will download the whole chat history.`);
                    return Promise.resolve(fbCount);
                }

                const diff = fbCount - dbCount;
                console.log(`Thread ${threadId}: Message cache is ${diff} messages behind`);

                return Promise.resolve(diff);
            })
            .then(msgCountToDownload => {
                const msgsPerRequest = Math.min(msgCountToDownload, 500);

                let i = 0;
                let lastTimestamp = +Date.now();
                let downloadedMessages = [];

                return new Promise((resolve, reject) => {
                    until(() => (i * msgsPerRequest) >= msgCountToDownload, (done) => {
                        this.api.getThreadHistory(threadId, 0, msgsPerRequest, lastTimestamp)
                            .then(messages => {
                                downloadedMessages = messages.concat(downloadedMessages);
                                lastTimestamp = parseInt(messages[0].timestamp) - 1;
                                i++;

                                console.log(`Thread ${threadId}: Downloaded total ${downloadedMessages.length}/${msgCountToDownload} messages (got ${messages.length})`);

                                done();
                            })
                            .catch(err => done(err));
                    }, (err) => {
                        if (err) return reject(err);

                        return resolve(downloadedMessages);
                    });
                });
            })
            .then(downloadedMessages => dbCollection.insertMany(downloadedMessages, {
                ordered: false // throw everything at the db and see what persists, i.e. don't stop on duplicate keys
            }))
            .catch(ThreadHistoryDownloadCancelledError, err => {
                return console.warn(err.message);
            })
            .catch(err => {
                // ignore duplicate keys error
                if (err.code === 11000) {
                    return console.warn(`Thread ${threadId}: There were one or more duplicate messages`);
                }

                throw err;
            });
    }
}

class ThreadHistoryDownloadCancelledError extends Error {
    constructor(msg) {
        super(msg);
    }
}<commit_msg>Refactor ThreadHistoryDownloader<commit_after>import { dependencies as Inject, singleton as Singleton } from "needlepoint";
import Db from "mongodb";
import ScheduledTask from "../ScheduledTask";
import Configuration from "../../core/config/Configuration";
import ChatApi from "../../core/api/ChatApi";
import Threads from "../../core/config/Threads";
import Promise from "bluebird";
import until from "async/until";

@Singleton
@Inject(Configuration, Db, ChatApi, Threads)
export default class ThreadHistoryDownloader extends ScheduledTask {
    /**
     * @param {Configuration} config
     * @param {Db} db
     * @param {ChatApi} api
     * @param {Threads} threads
     */
    constructor(config, db, api, threads) {
        super();

        this.config = config;
        this.db = db;
        this.api = api;
        this.threads = threads;
    }

    execute() {
        const threadIds = this.threads.getThreadIds();

        const promises = threadIds.map(id => this.downloadMessagesByThreadId(id));

        return Promise.all(promises);
    }

    downloadMessagesByThreadId(threadId) {
        const dbCollection = this.db.collection(`thread-${threadId}`);

        return Promise
            .all([
                ThreadHistoryDownloader.createUniqueIndexIfNotExists(dbCollection),
                dbCollection.count(),
                this.api.getThreadInfo(threadId).then(info => info.messageCount)
            ])
            .then(([_, dbCount, fbCount]) => ThreadHistoryDownloader.handleMessageCount(dbCount, fbCount, threadId))
            .then(messageCount => this.batchDownloadMessages(messageCount, threadId))
            .then(messages => dbCollection.insertMany(messages, {
                ordered: false // throw everything at the db and see what persists, i.e. don't stop on duplicate keys
            }))
            .catch(ThreadHistoryDownloadCancelledError, err => console.warn(err.message))
            .catch(err => ThreadHistoryDownloader.handleDuplicateKeysError(err, threadId));
    }

    batchDownloadMessages(messageCount, threadId) {
        const msgsPerRequest = Math.min(messageCount, 500);

        let i = 0;
        let lastTimestamp = +Date.now();
        let downloadedMessages = [];

        return new Promise((resolve, reject) => {
            until(
                () => (i * msgsPerRequest) >= messageCount,
                done => {
                    this.api.getThreadHistory(threadId, 0, msgsPerRequest, lastTimestamp).then(messages => {
                        downloadedMessages = messages.concat(downloadedMessages);
                        lastTimestamp = parseInt(messages[0].timestamp) - 1;
                        i++;

                        console.log(`Thread ${threadId}: Downloaded total ${downloadedMessages.length}/${messageCount} messages (got ${messages.length})`);

                        done();
                    }).catch(err => done(err));
                },
                err => {
                    if (err) return reject(err);

                    return resolve(downloadedMessages);
                });
        });
    }

    static handleMessageCount(dbCount, fbCount, threadId) {
        console.log(`Thread ${threadId}: There are ${dbCount} messages in cache, and ${fbCount} reported by Facebook`);

        if (dbCount == fbCount) {
            return Promise.reject(new ThreadHistoryDownloadCancelledError(
                `Thread ${threadId}: Message cache is up-to-date`
            ));
        }

        if (dbCount > fbCount) {
            console.warn(`There are more messages in the cache (${dbCount}) than Facebook reports (${fbCount}). ` +
                `Will download the whole chat history.`);
            return Promise.resolve(fbCount);
        }

        const diff = fbCount - dbCount;
        console.log(`Thread ${threadId}: Message cache is ${diff} messages behind`);

        return Promise.resolve(diff);
    }

    static createUniqueIndexIfNotExists(dbCollection) {
        const msgIdIndexName = "messageID";
        const uniqueMsgIdIndex = {};
        uniqueMsgIdIndex[msgIdIndexName] = 1;

        return dbCollection.indexExists(msgIdIndexName)
            .then(() => Promise.resolve())
            .catch(err => dbCollection.createIndex(uniqueMsgIdIndex, { unique: true }));
    }

    static handleDuplicateKeysError(err, threadId) {
        if (err.code === 11000) {
            return console.warn(`Thread ${threadId}: There were one or more duplicate messages`);
        }

        throw err;
    }
}

class ThreadHistoryDownloadCancelledError extends Error {
    constructor(msg) {
        super(msg);
    }
}
{'fraction_non_alphanumeric': 0.11213762344695763, 'fraction_numerical': 0.002760964213656154, 'mean_word_length': 1.9936427209154481, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5522997', 'n_tokens_mistral': 2419, 'n_tokens_neox': 2294, 'n_words': 574}