im.doctor/src/server/models/migration/migration.topics.js

/**
 * 群组，即原先的行政团队数据迁移。
 *
 * @author sand
 * @since 2017/1/9
 */
'use strict';

let ImDb = require('../../repository/oracle/db/im.db');
let ObjectUtil = require("../../util/object.util.js");
let DbUtil = require("../../util/db.util.js");

let async = require("async");
let log = require("../../util/log.js");
let fs = require('fs');
let mongoose = require("mongoose");
let vprintf = require('sprintf-js').vsprintf;

const MIGRATION_SCRIPT_File_NAME = "./ichat_1.2.8_topics_migration.sql";

class Migration {
    constructor() {
    }

    /**
     * 迁移议题及会话。
     */
    static migrateTopicAndSession() {
        let sessionIds = [];

        let data = "-- Topics update: \n";
        async.waterfall([
                /*function (callback) {
                 // 未导入的历史咨询
                 let sql = "select t.consult id, concat(t.patient, '_consult_2') session_id, t.name name, t.czrq create_time, t.end_operator end_by, t.end_time end_time, 0 'status', t.symptoms description " +
                 "from wlyy.wlyy_consult_team t where t.consult not in (select id from im.topics)";
                 ImDb.execQuery({
                 sql: sql,
                 args: [],
                 handler: function (err, res) {
                 if (err) {
                 return callback(err, res);
                 }

                 let buffer = "insert into topics(id, session_id, name, create_time, end_by, end_time, status, description) values ";
                 res.forEach(function (topic) {
                 buffer += vprintf("\n('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s'),", [
                 topic.id,
                 topic.session_id,
                 topic.name,
                 topic.create_time,
                 topic.end_by,
                 topic.end_time,
                 topic.status,
                 topic.description
                 ]);

                 topics.push(topic.id);
                 });

                 data += buffer.substr(0, buffer.length - 1) + " ON DUPLICATE KEY UPDATE id = id;\n";

                 callback(null);
                 }
                 });
                 },
                 function (callback) {
                 // 未导入的历史咨询所属会话
                 let sql = "select concat(c.patient, '_consult_2') id, c.name, 1 'type', 2 'business_type', c.czrq 'create_date' " +
                 "from wlyy.wlyy_consult_team c where concat(c.patient, '_consult_2') not in (select id from im.sessions s) group by c.patient order by c.patient";
                 ImDb.execQuery({
                 sql: sql,
                 args: [],
                 handler: function (err, res) {
                 if (err) {
                 return callback(err, res);
                 }

                 let buffer = "insert into sessions(id, name, type, business_type, create_date) values";
                 res.forEach(function (session) {
                 buffer += vprintf("\n('%s', '%s', '%s', '%s', '%s'),", [
                 session.id,
                 session.name,
                 session.type,
                 session.business_type,
                 session.create_date]);
                 });

                 data += buffer.substr(0, buffer.length - 1) + " ON DUPLICATE KEY UPDATE id = id;\n\n";

                 callback(null, null);
                 }
                 });
                 },*/
                function (callback) {
                    // 更新与未导入咨询相关的始末消息ID
                    let sql = "select t.id, t.session_id, t.name, t.create_time, t.end_by " +
                        "from topics t " +
                        "where t.start_message_id is null and t.end_message_id is null group by t.session_id " +
                        "order by session_id";
                    ImDb.execQuery({
                        sql: sql,
                        args: [],
                        handler: function (err, res) {
                            res.forEach(function (topic) {
                                let sessionId = topic.session_id;
                                sessionIds.push(sessionId);
                            });

                            callback(null);
                        }
                    });
                },
                function (callback) {
                    ImDb.execQuery({
                        sql: "select session_id, id message_id from muc_messages m where m.session_id in (?) order by session_id, m.id",
                        args: [sessionIds],
                        handler: function (err, res) {
                            let lastSessionId = null;
                            let lastMsgId = null;

                            let minMsgId = null;
                            let maxMsgId = null;

                            for(let i = 0; i < res.length; ++i){
                                console.log("Processing item " + i);
                                let item = res[i];

                                if(lastSessionId !== item.session_id){
                                    lastSessionId = item.session_id;

                                    maxMsgId = lastMsgId;
                                    if(maxMsgId !== null){
                                        let updateSQL = "update topics t set t.start_message_id = '%s', t.end_message_id = '%s' where t.session_id = '%s'; \n";
                                        updateSQL = vprintf(updateSQL, [minMsgId, maxMsgId, lastSessionId]);
                                        data += updateSQL;
                                    }

                                    minMsgId = item.message_id;
                                }

                                lastMsgId = item.message_id;

                                if(i === res.length - 1){
                                    let updateSQL = "update topics t set t.start_message_id = '%s', t.end_message_id = '%s' where t.session_id = '%s'; \n";
                                    updateSQL = vprintf(updateSQL, [minMsgId, lastMsgId, lastSessionId]);
                                    data += updateSQL;

                                    callback(null, null);
                                }
                            }
                        }
                    });
                }],
            function (err, res) {
                if (err) {
                    log.error("Error occurs while migrating topics and sessions: ", err);
                } else {
                    fs.writeFileSync(MIGRATION_SCRIPT_File_NAME, data + "\n");

                    log.info("Migrate topics and sessions succeed");
                }
            });
    }
}

async.waterfall([
        function (callback) {
            Migration.migrateTopicAndSession();
            callback(null);
        }
    ],
    function (err, res) {

    });