liberama/server/core/Reader/ReaderWorker.js

const fs = require('fs-extra');
const path = require('path');

const LimitedQueue = require('../LimitedQueue');
const WorkerState = require('../WorkerState');//singleton
const FileDownloader = require('../FileDownloader');
const FileDecompressor = require('../FileDecompressor');
const BookConverter = require('./BookConverter');
const RemoteStorage = require('../RemoteStorage');
const JembaConnManager = require('../../db/JembaConnManager');//singleton
const ayncExit = new (require('../AsyncExit'))();

const utils = require('../utils');
const log = new (require('../AppLogger'))().log;//singleton

const cleanDirPeriod = 60*60*1000;//каждый час
const remoteSendPeriod = 119*1000;//примерно раз 2 минуты

const queue = new LimitedQueue(5, 100, 2*60*1000 + 15000);//2 минуты ожидание подвижек

let instance = null;

//singleton
class ReaderWorker {
    constructor(config) {
        if (!instance) {
            this.config = Object.assign({}, config);
            
            this.config.tempDownloadDir = `${config.tempDir}/download`;
            fs.ensureDirSync(this.config.tempDownloadDir);

            this.config.tempPublicDir = `${config.publicDir}/tmp`;
            fs.ensureDirSync(this.config.tempPublicDir);

            this.workerState = new WorkerState();
            this.down = new FileDownloader(config.maxUploadFileSize);
            this.decomp = new FileDecompressor(3*config.maxUploadFileSize);
            this.bookConverter = new BookConverter(this.config);

            this.connManager = new JembaConnManager();
            this.appDb = this.connManager.db['app'];

            this.remoteStorage = false;
            if (config.remoteStorage) {
                this.remoteStorage = new RemoteStorage(
                    Object.assign({maxContentLength: 3*config.maxUploadFileSize}, config.remoteStorage)
                );
            }

            this.dirConfigArr = [
                {
                    dir: this.config.tempPublicDir,
                    remoteDir: '/tmp',
                    maxSize: this.config.maxTempPublicDirSize,
                    moveToRemote: true,
                },
                {
                    dir: this.config.uploadDir,
                    remoteDir: '/upload',
                    maxSize: this.config.maxUploadPublicDirSize,
                    moveToRemote: true,
                }
            ];
            //преобразуем в объект для большего удобства
            this.dirConfig = {};
            for (const configRec of this.dirConfigArr)
                this.dirConfig[configRec.remoteDir] = configRec;

            this.remoteFilesToSend = [];
            this.periodicCleanDir();//no await
            
            instance = this;
        }

        return instance;
    }

    async loadBook(opts, wState) {
        const url = opts.url;
        let decompDir = '';
        let downloadedFilename = '';
        let isUploaded = false;
        let convertFilename = '';

        const overLoadMes = 'Слишком большая очередь загрузки. Пожалуйста, попробуйте позже.';
        const overLoadErr = new Error(overLoadMes);

        let q = null;
        try {
            wState.set({state: 'queue', step: 1, totalSteps: 1});
            try {
                let qSize = 0;
                q = await queue.get((place) => {
                    wState.set({place, progress: (qSize ? Math.round((qSize - place)/qSize*100) : 0)});
                    if (!qSize)
                        qSize = place;
                });
            } catch (e) {
                throw overLoadErr;
            }

            wState.set({state: 'download', step: 1, totalSteps: 3, url});

            const tempFilename = utils.randomHexString(30);
            const tempFilename2 = utils.randomHexString(30);
            const decompDirname = utils.randomHexString(30);

            let downloadSize = -1;
            //download or use uploaded
            if (url.indexOf('disk://') != 0) {//download
                const downdata = await this.down.load(url, (progress) => {
                    wState.set({progress});
                }, q.abort);

                downloadedFilename = `${this.config.tempDownloadDir}/${tempFilename}`;

                downloadSize = downdata.length;
                await fs.writeFile(downloadedFilename, downdata);
            } else {//uploaded file
                const fileHash = url.substr(7);
                downloadedFilename = `${this.config.uploadDir}/${fileHash}`;
                if (!await fs.pathExists(downloadedFilename)) {
                    //если удалено из upload, попробуем восстановить из удаленного хранилища
                    try {
                        await this.restoreRemoteFile(fileHash, '/upload');
                    } catch(e) {
                        throw new Error('Файл не найден на сервере (возможно был удален как устаревший). Пожалуйста, загрузите файл с диска на сервер заново.');
                    }
                }
                await utils.touchFile(downloadedFilename);
                isUploaded = true;
            }
            wState.set({progress: 100});

            if (q.abort())
                throw overLoadErr;
            q.resetTimeout();

            //decompress
            wState.set({state: 'decompress', step: 2, progress: 0});
            decompDir = `${this.config.tempDownloadDir}/${decompDirname}`;
            let decompFiles = {};
            try {
                decompFiles = await this.decomp.decompressNested(downloadedFilename, decompDir);
            } catch (e) {
                log(LM_ERR, e.stack);
                throw new Error('Ошибка распаковки');
            }
            wState.set({progress: 100});
            
            if (q.abort())
                throw overLoadErr;
            q.resetTimeout();

            //конвертирование в fb2
            wState.set({state: 'convert', step: 3, progress: 0});
            convertFilename = `${this.config.tempDownloadDir}/${tempFilename2}`;
            await this.bookConverter.convertToFb2(decompFiles, convertFilename, opts, progress => {
                wState.set({progress});
                if (queue.freed > 0)
                    q.resetTimeout();
            }, q.abort);

            //сжимаем файл в tmp, если там уже нет с тем же именем-sha256
            const compFilename = await this.decomp.gzipFileIfNotExists(convertFilename, this.config.tempPublicDir);
            const stat = await fs.stat(compFilename);

            wState.set({progress: 100});

            //finish
            const finishFilename = path.basename(compFilename);

            const result = {path: `/tmp/${finishFilename}`, size: stat.size};
            if (downloadSize >= 0)
                result.downloadSize = downloadSize;

            wState.finish(result);

            //асинхронно через 30 сек добавим в очередь на отправку
            //т.к. gzipFileIfNotExists может переупаковать файл
            (async() => {
                await utils.sleep(30*1000);
                this.pushRemoteSend(compFilename, '/tmp');
            })();

        } catch (e) {
            log(LM_ERR, e.stack);
            let mes = e.message.split('|FORLOG|');
            if (mes[1])
                log(LM_ERR, mes[0] + mes[1]);
            log(LM_ERR, `downloadedFilename: ${downloadedFilename}`);

            mes = mes[0];
            if (mes == 'abort')
                mes = overLoadMes;
            wState.set({state: 'error', error: mes});
        } finally {
            //clean
            if (q)
                q.ret();
            if (decompDir)
                await fs.remove(decompDir);
            if (downloadedFilename && !isUploaded)
                await fs.remove(downloadedFilename);
            if (convertFilename)
                await fs.remove(convertFilename);
        }
    }

    loadBookUrl(opts) {
        const workerId = this.workerState.generateWorkerId();
        const wState = this.workerState.getControl(workerId);
        wState.set({state: 'start'});

        this.loadBook(opts, wState);

        return workerId;
    }

    async saveFile(file) {
        const hash = await utils.getFileHash(file.path, 'sha256', 'hex');
        const outFilename = `${this.config.uploadDir}/${hash}`;

        if (!await fs.pathExists(outFilename)) {
            await fs.move(file.path, outFilename);
            this.pushRemoteSend(outFilename, '/upload');
        } else {
            await utils.touchFile(outFilename);
            await fs.remove(file.path);
        }

        return `disk://${hash}`;
    }

    async saveFileBuf(buf) {
        const hash = await utils.getBufHash(buf, 'sha256', 'hex');
        const outFilename = `${this.config.uploadDir}/${hash}`;

        if (!await fs.pathExists(outFilename)) {
            await fs.writeFile(outFilename, buf);
            this.pushRemoteSend(outFilename, '/upload');
        } else {
            await utils.touchFile(outFilename);
        }

        return `disk://${hash}`;
    }

    async uploadFileTouch(url) {
        const outFilename = `${this.config.uploadDir}/${url.replace('disk://', '')}`;

        await utils.touchFile(outFilename);

        return url;
    }

    async restoreRemoteFile(filename, remoteDir) {
        let targetDir = '';
        if (this.dirConfig[remoteDir])
            targetDir = this.dirConfig[remoteDir].dir;
        else
            throw new Error(`restoreRemoteFile: unknown remoteDir value (${remoteDir})`);

        const basename = path.basename(filename);
        const targetName = `${targetDir}/${basename}`;

        if (!await fs.pathExists(targetName)) {
            let found = false;
            if (this.remoteStorage) {
                found = await this.remoteStorage.getFileSuccess(targetName, remoteDir);
            }

            if (!found) {
                throw new Error('404 Файл не найден');
            }
        }

        return targetName;
    }

    pushRemoteSend(fileName, remoteDir) {
        if (this.remoteStorage
            && this.dirConfig[remoteDir] 
            && this.dirConfig[remoteDir].moveToRemote) {
            this.remoteFilesToSend.push({fileName, remoteDir});
        }
    }

    async remoteSendFile(sendFileRec) {
        const {fileName, remoteDir} = sendFileRec;
        const sent = this.remoteSent;

        if (!fileName || sent[fileName])
            return;

        log(`remoteSendFile ${remoteDir}/${path.basename(fileName)}`);

        //отправляем в remoteStorage
        await this.remoteStorage.putFile(fileName, remoteDir);
        
        sent[fileName] = true;
        await this.appDb.insert({table: 'remote_sent', ignore: true, rows: [{id: fileName, remoteDir}]});
    }

    async remoteSendAll() {
        if (!this.remoteStorage)
            return;

        const newSendQueue = [];
        while (this.remoteFilesToSend.length) {
            const sendFileRec = this.remoteFilesToSend.shift();

            if (sendFileRec.remoteDir
                && this.dirConfig[sendFileRec.remoteDir]
                && this.dirConfig[sendFileRec.remoteDir].moveToRemote) {

                try {
                    await this.remoteSendFile(sendFileRec);
                } catch (e) {
                    newSendQueue.push(sendFileRec)
                    log(LM_ERR, e.stack);
                }
            }
        }

        this.remoteFilesToSend = newSendQueue;
    }

    async cleanDir(config) {
        const {dir, remoteDir, maxSize, moveToRemote} = config;
        const sent = this.remoteSent;

        const list = await fs.readdir(dir);

        let size = 0;
        let files = [];
        for (const filename of list) {
            const filePath = `${dir}/${filename}`;
            const stat = await fs.stat(filePath);
            if (!stat.isDirectory()) {
                size += stat.size;
                files.push({name: filePath, stat});
            }
        }

        log(LM_WARN, `clean dir ${dir}, maxSize=${maxSize}, found ${files.length} files, total size=${size}`);

        files.sort((a, b) => a.stat.mtimeMs - b.stat.mtimeMs);

        //удаленное хранилище
        if (moveToRemote && this.remoteStorage) {
            const foundFiles = new Set();
            for (const file of files) {
                foundFiles.add(file.name);

                //отсылаем на всякий случай перед удалением, если вдруг remoteSendAll не справился
                try {
                    await this.remoteSendFile({fileName: file.name, remoteDir});
                } catch (e) {
                    log(LM_ERR, e.stack);
                }
            }

            //почистим remoteSent и БД
            //несколько неоптимально, таскает все записи из таблицы
            const rows = await this.appDb.select({table: 'remote_sent'});
            for (const row of rows) {
                if ((row.remoteDir === remoteDir && !foundFiles.has(row.id))
                    || !this.dirConfig[row.remoteDir]) {
                    delete sent[row.id];
                    await this.appDb.delete({table: 'remote_sent', where: `@@id(${this.appDb.esc(row.id)})`});
                }
            }
        }

        let i = 0;
        let j = 0;
        while (i < files.length && size > maxSize) {
            const file = files[i];
            const oldFile = file.name;

            //реально удаляем только если сохранили в хранилище или размер dir увеличен в 1.5 раза
            if (!(moveToRemote && this.remoteStorage)
                || (moveToRemote && this.remoteStorage && sent[oldFile])
                || size > maxSize*1.5) {
                await fs.remove(oldFile);
                j++;
            }
            
            size -= file.stat.size;
            i++;
        }

        log(LM_WARN, `removed ${j} files`);
    }

    async periodicCleanDir() {
        try {
            if (!this.remoteSent)
                this.remoteSent = {};

            //инициализация this.remoteSent
            if (this.remoteStorage) {
                const rows = await this.appDb.select({table: 'remote_sent'});
                for (const row of rows) {
                    this.remoteSent[row.id] = true;
                }
            }

            let lastCleanDirTime = 0;
            let lastRemoteSendTime = 0;
            while (1) {// eslint-disable-line no-constant-condition
                //отсылка в удаленное хранилище
                if (Date.now() - lastRemoteSendTime >= remoteSendPeriod) {
                    try {
                        await this.remoteSendAll();
                    } catch(e) {
                        log(LM_ERR, e.stack);
                    }

                    lastRemoteSendTime = Date.now();
                }

                //чистка папок
                if (Date.now() - lastCleanDirTime >= cleanDirPeriod) {
                    for (const config of Object.values(this.dirConfig)) {
                        try {
                            await this.cleanDir(config);
                        } catch(e) {
                            log(LM_ERR, e.stack);
                        }
                    }

                    lastCleanDirTime = Date.now();
                }

                await utils.sleep(60*1000);//интервал проверки 1 минута
            }
        } catch (e) {
            log(LM_FATAL, e.message);
            ayncExit.exit(1);
        }
    }

}

module.exports = ReaderWorker;