tutuxiang-desktop/src/service/FileUploadDownloadService.js

const fs = require('fs');
const path = require('path');
const app = require('electron').app || require('electron').remote.app;
const { UploadStatus, DownloadStatus } = require('../Constants');
const OSS = require('ali-oss');
const readChunk = require('read-chunk');
const fileType = require('file-type');
const ImageUtils = require('./ImageUtils').default;
const { groupBy } = require('lodash');
const archiver = require('archiver');
const uuidv1 = require('uuid/v1');
const eventBus = require('./EventBus').default;
const queue = require('block-queue');
const Sequelize = require('sequelize');
const Op = Sequelize.Op;
const { UploadJob } = require('./UploadJob');
const { DownloadJob } = require('./DownloadJob');
const { Order, SubOrder, File, Download } = require('./model');
const rimraf = require('rimraf');
const fsUtils = require('nodejs-fs-utils');

if (!fs.existsSync(path.resolve(app.getPath('userData'), 'db'))) {
    fs.mkdirSync(path.resolve(app.getPath('userData'), 'db'));
}
let dbDir = path.resolve(app.getPath('userData'), 'db');
let archiveDir = path.resolve(app.getPath('userData'), 'archive');
let thumbDir = path.resolve(app.getPath('userData'), 'thumb');
console.log(dbDir);

if (!fs.existsSync(archiveDir)) {
    fs.mkdirSync(archiveDir);
}

const sequelize = new Sequelize({
    dialect: 'sqlite',
    storage: path.resolve(app.getPath('userData'), 'db', 'application.db'),
    logging: false,
});

const Model = Sequelize.Model;
Order.init(sequelize);
SubOrder.init(sequelize);
File.init(sequelize);
Download.init(sequelize);

let db = {
    Order,
    SubOrder,
    File,
    Download,
};
window.db = db;

sequelize.sync().then(() => {
    checkFalse();
    init();
});

let uploadQueue = queue(3, function(job, done) {
    job.start(done);
});

//下载队列
let downloadQueue = queue(3, function(job, done) {
    job.start(done);
});

//检查错误下载或上传
async function checkFalse() {
    //下载,下载队列为空,把所有状态为downloading的状态改为ERROR
    let downloadList = (await db.Download.findAll({
        where: { status: DownloadStatus.DOWNLOADING, status: DownloadStatus.WAITING },
    })).map(i => i.get());
    downloadList.forEach(async i => {
        await db.Download.update(
            {
                status: DownloadStatus.PAUSE,
            },
            {
                where: {
                    id: i.id,
                },
            },
        );
    });

    //上传
    let fileList = (await db.File.findAll({ where: { status: UploadStatus.UPLOADING } })).map(i =>
        i.get(),
    );
    fileList.forEach(async i => {
        await db.File.update(
            {
                status: UploadStatus.PAUSE,
            },
            {
                where: {
                    id: i.id,
                },
            },
        );
    });
}

function sleep(ms) {
    return new Promise(resolve => {
        setTimeout(resolve, ms);
    });
}

let initiated = false;
const init = async () => {
    if (initiated) {
        return;
    }
    resume();

    initiated = true;
};

const PromiseQueue = arr => {
    var sequence = Promise.resolve();
    arr.forEach(function(item) {
        sequence = sequence.then(item).then(() => {
            return res;
        });
    });
    return sequence;
};

const pause = async arg => {
    if (arg && arg.orderId) {
        await db.Order.update(
            {
                pause: 1,
            },
            {
                where: {
                    id: arg.orderId,
                },
            },
        );
        await db.SubOrder.update(
            { pause: 1 },
            {
                where: {
                    orderId: arg.orderId,
                },
            },
        );
    }
    if (arg && arg.subOrderId) {
        await db.SubOrder.update(
            { pause: 1 },
            {
                where: {
                    id: arg.subOrderId,
                },
            },
        );
    }
    eventBus.publish('pause', arg);
};

const resume = async arg => {
    let records = [];
    if (arg && arg.orderId) {
        records = (await db.File.findAll({
            where: {
                orderId: arg.orderId,
                status: {
                    [Op.not]: UploadStatus.FINISHED,
                },
            },
        })).map(i => i.get());
    } else if (arg && arg.subOrderId) {
        records = (await db.File.findAll({
            where: {
                subOrderId: arg.subOrderId,
                status: {
                    [Op.not]: UploadStatus.FINISHED,
                },
            },
        })).map(i => i.get());
    } else if (arg && arg.fileId) {
        records = (await db.File.findAll({
            where: {
                id: arg.fileId,
                status: {
                    [Op.not]: UploadStatus.FINISHED,
                },
            },
        })).map(i => i.get());
    } else {
        records = (await db.File.findAll({
            where: {
                status: {
                    [Op.not]: UploadStatus.FINISHED,
                },
            },
        })).map(i => i.get());
    }
    let orderIds = [];
    let subOrderIds = [];
    records.forEach(i => {
        if (orderIds.indexOf(i.orderId) === -1) {
            orderIds.push(i.orderId);
        }
        if (subOrderIds.indexOf(i.subOrderId) === -1) {
            subOrderIds.push(i.subOrderId);
        }
    });
    await db.Order.update(
        {
            pause: 0,
        },
        {
            where: {
                id: orderIds,
            },
        },
    );
    await db.SubOrder.update(
        { pause: 0 },
        {
            where: {
                id: subOrderIds,
            },
        },
    );
    records.forEach(i => {
        let job = new UploadJob(i, db);
        uploadQueue.push(job);
    });
};

const queryUnfinishedOrder = async userId => {
    // let sql = `
    //     SELECT
    //         o.*,
    //         fileNum,
    //         totalSize,
    //         uploadedSize
    //     FROM
    //         Orders o
    //         JOIN ( SELECT SUM( size ) AS totalSize, SUM( progress * size ) AS uploadedSize, COUNT( * ) AS fileNum, orderId  FROM Files GROUP BY orderId ) t ON o.id = t.orderId
    //     WHERE
    //         o.id IN (
    //         SELECT
    //             orderId
    //         FROM
    //             Files
    //         WHERE
    //             status != ${UploadStatus.FINISHED}
    //     GROUP BY
    //         orderId)
    //         and o.userId = ${userId}`;
    let sql = `
        SELECT
            Orders.* ,
            ( SELECT SUM( size ) FROM SubOrders WHERE SubOrders.orderId = Orders.id ) AS totalSize,
            ( SELECT SUM( f0.progress * f0.size ) FROM Files f0 WHERE f0.orderId = Orders.id ) AS uploadedSize,
            ( SELECT COUNT( f1.id ) FROM Files f1 WHERE f1.orderId = Orders.id ) AS fileNum 
        FROM
            Orders
            JOIN Files ON Files.orderId = Orders.id 
        WHERE
            Files.status != ${UploadStatus.FINISHED}
            AND Orders.userId = ${userId}
        GROUP BY
            Orders.id
    `;
    let res = await sequelize.query(sql, { type: sequelize.QueryTypes.SELECT });
    return res;
};

const queryFinishedOrder = async userId => {
    let res = await sequelize.query(
        `SELECT
        *,
        ( SELECT COUNT( * ) FROM Files WHERE Files.orderId = Orders.id ) AS totalNum,
        ( SELECT COUNT( * ) FROM Files WHERE Files.orderId = Orders.id AND status = ${
            UploadStatus.FINISHED
        } ) AS finishedNum,
        ( SELECT SUM( size ) FROM Files WHERE Files.orderId = Orders.id ) AS totalSize 
    FROM
        Orders 
    WHERE
    totalNum = finishedNum AND userId= ${userId}
    order by updatedAt DESC`,
        { type: sequelize.QueryTypes.SELECT },
    );
    return res;
};

const queryDetail = async orderId => {
    let res = await sequelize.query(
        `SELECT
            s.id,
            s.orderId,
            s.parentId,
            s.path,
            s.name,
            s.size AS totalSize,
            s.pause,
            s.createdAt,
            s.updatedAt,
            COUNT( * ) AS fileNum,
            ( SELECT COUNT( * ) FROM Files WHERE src LIKE '%.zip' AND subOrderId = s.id ) > 0 AS hasZip,
            ( SELECT COUNT( * ) FROM Files WHERE status = - 1 AND subOrderId = s.id ) > 0 AS packing,
            ( SUM( progress * f.size ) / s.size) AS progress,
            (
            SELECT
                a = b 
            FROM
                (
                    ( SELECT COUNT( * ) AS a FROM Files ta WHERE ta.subOrderId = f.subOrderId )
                    JOIN ( SELECT COUNT( * ) AS b FROM Files tb WHERE tb.subOrderId = f.subOrderId AND (tb.status = ${
                        UploadStatus.PAUSE
                    } OR tb.status = ${UploadStatus.FINISHED}) ) 
                ) 
            ) AS paused
        FROM
            Files f
            JOIN SubOrders s ON f.subOrderId = s.id 
        WHERE
            f.orderId = ${orderId} 
        GROUP BY
            f.subOrderId`,
        { type: sequelize.QueryTypes.SELECT },
    );
    return res;
};

const transferFilename = filename => {
    return filename.replace(/\.[0-9a-z]+$/i, '') + '.jpg';
};

const checkClientOrder = async clientOrderId => {
    let res = await sequelize.query(
        `SELECT 
            * 
        FROM 
            Orders 
        WHERE 
        clientOrderId = ${clientOrderId} `,
        { type: sequelize.QueryTypes.SELECT },
    );
    return res;
};

const uploadOrder = async (topDir, prefixes, userId, clientOrderId) => {
    console.log(topDir, JSON.stringify(prefixes), userId, clientOrderId);
    if (!prefixes instanceof Object) {
        throw 'prefixes not found';
    }
    let name = path.basename(topDir);
    let mainOrder = (await db.Order.create({
        path: topDir,
        name: name,
        userId: userId,
        clientOrderId: clientOrderId,
        pause: 0,
    })).get();
    let orderId = mainOrder.id;

    let hasFile = false;
    let hasSubDir = false;
    fs.readdirSync(topDir).forEach(child => {
        if (!child.startsWith('.')) {
            let info = fs.statSync(path.resolve(topDir, child));
            if (info.isDirectory()) {
                hasSubDir = true;
            } else {
                hasFile = true;
            }
        }
    });
    if (hasFile && hasSubDir) {
        throw '不能同时包含文件夹和文件';
    }
    if (!hasFile && !hasSubDir) {
        throw '空文件夹';
    }
    if (!hasSubDir) {
        if (prefixes[0]) {
            throw 'prefixes not found';
        }
        let prefix = prefixes[path.basename(topDir)];
        let subOrder = (await db.SubOrder.create({
            orderId,
            parentId: mainOrder.id,
            path: topDir,
            name,
            size: fsUtils.fsizeSync(topDir),
            pause: 0,
        })).get();

        let outPath = path.resolve(archiveDir, uuidv1() + '.zip');
        let fileinfo = (await db.File.create({
            orderId,
            subOrderId: subOrder.id,
            src: outPath,
            // dst: orderId + "/" + path.basename(outPath),
            dst: prefix + '/' + path.basename(topDir) + '_original.zip',
            size: 1,
            uploaded: 0,
            progress: 0,
            status: UploadStatus.PREPARING,
        })).get();
        await readDir(topDir, orderId, subOrder.id, null, prefix);
        let archiveFile = await archive(topDir, outPath);
        let info = fs.statSync(archiveFile);
        await sequelize.query(`UPDATE SubOrders 
            SET size = ( SELECT SUM( size ) FROM Files WHERE subOrderId = ${subOrder.id} ) 
            WHERE id = ${subOrder.id}`);
        await db.File.update(
            {
                status: UploadStatus.WAITING,
                size: info.size,
            },
            { where: { id: fileinfo.id } },
        );
        uploadQueue.push(new UploadJob(fileinfo, db));
    } else {
        fs.readdirSync(topDir).forEach(async child => {
            if (!child.startsWith('.')) {
                let prefix = prefixes[child];
                let info = fs.statSync(path.resolve(topDir, child));
                if (info.isDirectory()) {
                    let subOrder = (await db.SubOrder.create({
                        orderId,
                        parentId: mainOrder.id,
                        path: path.resolve(topDir, child),
                        name: child,
                        size: fsUtils.fsizeSync(path.resolve(topDir, child)),
                        pause: 0,
                    })).get();

                    let outPath = path.resolve(archiveDir, uuidv1() + '.zip');
                    let fileinfo = (await db.File.create({
                        orderId,
                        subOrderId: subOrder.id,
                        src: outPath,
                        // dst: orderId + "/" + path.basename(outPath),
                        dst: prefix + '/' + child + '_original.zip',
                        size: 1,
                        uploaded: 0,
                        progress: 0,
                        status: UploadStatus.PREPARING,
                    })).get();
                    await readDir(path.resolve(topDir, child), orderId, subOrder.id, child, prefix);
                    let archiveFile = await archive(path.resolve(topDir, child), outPath);
                    let info = fs.statSync(archiveFile);
                    await db.File.update(
                        {
                            status: UploadStatus.WAITING,
                            size: info.size,
                        },
                        { where: { id: fileinfo.id } },
                    );
                    await sequelize.query(`
                    UPDATE SubOrders SET size = ( SELECT SUM( size ) FROM Files WHERE subOrderId = ${
                        subOrder.id
                    } ) WHERE id = ${subOrder.id}`);
                    uploadQueue.push(new UploadJob(fileinfo, db));
                }
            }
        });
    }
    async function readDir(dir, orderId, subOrderId, relative, prefix) {
        for await (const child of fs.readdirSync(dir)) {
            if (!child.startsWith('.')) {
                let currRelative = relative ? relative + '/' + child : child;
                let info = fs.statSync(path.resolve(dir, child));
                if (info.isDirectory()) {
                    await readDir(
                        path.resolve(dir, child),
                        orderId,
                        subOrderId,
                        currRelative,
                        prefix,
                    );
                } else {
                    let filepath = path.resolve(dir, child);
                    const buffer = readChunk.sync(filepath, 0, fileType.minimumBytes);
                    let type = fileType(buffer);
                    let src = filepath;
                    let dst = currRelative;
                    if (type && type.mime.startsWith('image')) {
                        src = await ImageUtils.makeThumbnail(filepath);
                        // dst = orderId + "/" + relative + '/' + transferFilename(path.basename(currRelative))
                        dst =
                            prefix +
                            '/original/min/' +
                            transferFilename(path.basename(currRelative));
                        info = fs.statSync(src);
                        console.log('thumbnail:' + src);
                        let fileinfo = (await db.File.create({
                            orderId,
                            subOrderId,
                            src,
                            dst,
                            size: info.size,
                            uploaded: 0,
                            progress: 0,
                            status: UploadStatus.WAITING,
                        })).get();
                        console.log(fileinfo);
                        uploadQueue.push(new UploadJob(fileinfo, db));
                    }
                }
            }
        }
    }
};

const archive = async (dir, outPath) => {
    return new Promise((resolve, reject) => {
        // create a file to stream archive data to.
        let output = fs.createWriteStream(outPath);
        let archive = archiver('zip', {
            zlib: { level: 9 }, // Sets the compression level.
        });

        // listen for all archive data to be written
        // 'close' event is fired only when a file descriptor is involved
        output.on('close', function() {
            console.log(archive.pointer() + ' total bytes');
            console.log('archiver has been finalized and the output file descriptor has closed.');
            resolve(outPath);
        });

        // This event is fired when the data source is drained no matter what was the data source.
        // It is not part of this library but rather from the NodeJS Stream API.
        // @see: https://nodejs.org/api/stream.html#stream_event_end
        output.on('end', function() {
            console.log('Data has been drained');
        });

        // good practice to catch warnings (ie stat failures and other non-blocking errors)
        archive.on('warning', function(err) {
            if (err.code === 'ENOENT') {
                // log warning
            } else {
                // throw error
                reject(err);
            }
        });

        // good practice to catch this error explicitly
        archive.on('error', function(err) {
            reject(err);
        });

        // pipe archive data to the file
        archive.pipe(output);
        archive.directory(dir, path.basename(dir));

        archive.finalize();
    });
};

const clearDB = async () => {
    await db.Order.destroy({ where: {} });
    await db.SubOrder.destroy({ where: {} });
    await db.File.destroy({ where: {} });
    require('child_process').execSync('rm -rf *', {
        cwd: path.resolve(app.getPath('userData'), 'thumb'),
    });
    require('child_process').execSync('rm -rf *', {
        cwd: path.resolve(app.getPath('userData'), 'archive'),
    });
    console.log('done');
};

//下载方法
const downloadPause = async arg => {
    console.log('暂停');
    eventBus.publish('downpause', arg);
};

const downloadResume = async arg => {
    if (arg.id) {
        await db.Download.update(
            { status: DownloadStatus.WAITING },
            {
                where: {
                    id: arg.id,
                },
            },
        );
        var record = await db.Download.findOne({ where: { id: arg.id } });

        console.log(record);
        let job = new DownloadJob(record, db);
        downloadQueue.push(job);
    }
};
const queryUnfinishedDownloadOrder = async userId => {
    let res = await sequelize.query(
        `SELECT 
            * 
        FROM 
            Downloads 
        WHERE 
            userId = ${userId} AND status != ${DownloadStatus.FINISHED}`,
        { type: sequelize.QueryTypes.SELECT },
    );
    return res;
};

const queryfinishedDownloadOrder = async userId => {
    let res = await sequelize.query(
        `SELECT 
            * 
        FROM 
            Downloads 
        WHERE 
            userId = ${userId} AND status = ${DownloadStatus.FINISHED}
        order by updatedAt DESC`,
        { type: sequelize.QueryTypes.SELECT },
    );
    return res;
};

async function getFilePath(pathName, i) {
    var _path = path.resolve(window.localStorage['downLoadingPath2'], pathName);
    var index = i ? i : 0;
    if (index) {
        _path = path.resolve(
            window.localStorage['downLoadingPath2'],
            path.basename(pathName, path.extname(pathName)) +
                '(' +
                i +
                ')' +
                path.extname(pathName),
        );
    }
    var exist = await fs.existsSync(_path);

    if (exist) {
        _path = await getFilePath(pathName, index + 1);
    }
    return _path;
}

const downloadFile = async (objectName, orderName, userId, quantity, userOrderId) => {
    let pathName = await getFilePath(userOrderId + '_' + orderName, 0);
    let download = (await db.Download.create({
        dst: objectName,
        size: 0,
        path: pathName,
        name: path.basename(pathName),
        progress: 0,
        checkPoint: 0,
        status: DownloadStatus.WAITING,
        userId: userId,
        finishedNum: quantity,
        userOrderId: userOrderId,
    })).get();
    downloadQueue.push(new DownloadJob(download, db));
};
const clearCache = async () => {
    rmDir = function(dirPath) {
        try {
            var files = fs.readdirSync(dirPath);
        } catch (e) {
            return;
        }
        for (var i = 0; i < files.length; i++) {
            var filePath = path.resolve(dirPath, files[i]);
            console.log(filePath);
            if (fs.statSync(filePath).isFile()) fs.unlinkSync(filePath);
            else rmDir(filePath);
        }
    };
    rmDir(path.resolve(app.getPath('userData'), 'archive'));
    rmdir(path.resolve(app.getPath('userData'), 'thumb'));
};
window.clearCache = clearCache;

const deleteUploadedFile = async () => {
    let files = await db.File.findAll({
        where: {
            status: UploadStatus.FINISHED,
        },
    }).map(i => i.get());
    files.forEach(i => {
        fs.unlink(i.src, err => {
            if (!err) {
                console.log(`${i.src} deleted`);
            }
        });
    });
};

export default {
    init,
    pause,
    uploadOrder,
    queryUnfinishedOrder,
    clearDB,
    archive,
    queryDetail,
    resume,
    queryFinishedOrder,
    downloadFile,
    downloadPause,
    downloadResume,
    queryUnfinishedDownloadOrder,
    queryfinishedDownloadOrder,
    checkClientOrder,
    deleteUploadedFile,
};