HackMD/lib/realtime.js

//realtime
//external modules
var cookie = require('cookie');
var cookieParser = require('cookie-parser');
var url = require('url');
var async = require('async');
var LZString = require('lz-string');
var shortId = require('shortid');
var randomcolor = require("randomcolor");
var Chance = require('chance'),
    chance = new Chance();

//core
var config = require("../config.js");

//others
var db = require("./db.js");
var Note = require("./note.js");
var User = require("./user.js");

//public
var realtime = {
    onAuthorizeSuccess: onAuthorizeSuccess,
    onAuthorizeFail: onAuthorizeFail,
    secure: secure,
    connection: connection,
    getStatus: getStatus
};

function onAuthorizeSuccess(data, accept) {
    accept(null, true);
}

function onAuthorizeFail(data, message, error, accept) {
    if (error) throw new Error(message);
    accept(null, true);
}

function secure(socket, next) {
    try {
        var handshakeData = socket.request;
        if (handshakeData.headers.cookie) {
            handshakeData.cookie = cookie.parse(handshakeData.headers.cookie);
            handshakeData.sessionID = cookieParser.signedCookie(handshakeData.cookie[config.sessionname], config.sessionsecret);
            if (handshakeData.cookie[config.sessionname] == handshakeData.sessionID) {
                next(new Error('AUTH failed: Cookie is invalid.'));
            }
        } else {
            next(new Error('AUTH failed: No cookie transmitted.'));
        }
        if (config.debug)
            console.log("AUTH success cookie: " + handshakeData.sessionID);

        next();
    } catch (ex) {
        next(new Error("AUTH failed:" + JSON.stringify(ex)));
    }
}

//actions
var users = {};
var notes = {};
var updater = setInterval(function () {
    async.each(Object.keys(notes), function (key, callback) {
        var note = notes[key];
        if (note.isDirty) {
            if (config.debug)
                console.log("updater found dirty note: " + key);
            var body = LZString.decompressFromUTF16(note.body);
            var title = Note.getNoteTitle(body);
            body = LZString.compressToBase64(body);
            db.saveToDB(key, title, body,
                function (err, result) {});
            note.isDirty = false;
        }
        callback();
    }, function (err) {
        if (err) return console.error('updater error', err);
    });
}, 5000);

function getStatus(callback) {
    db.countFromDB(function (err, data) {
        if (err) return console.log(err);
        var regusers = 0;
        var distinctregusers = 0;
        var distinctaddresses = [];
        Object.keys(users).forEach(function (key) {
            var value = users[key];
            if (value.login)
                regusers++;
            var found = false;
            for (var i = 0; i < distinctaddresses.length; i++) {
                if (value.address == distinctaddresses[i]) {
                    found = true;
                    break;
                }
            }
            if (!found) {
                distinctaddresses.push(value.address);
                if (value.login)
                    distinctregusers++;
            }
        });
        User.getUserCount(function (err, regcount) {
            if (err) {
                console.log('get status failed: ' + err);
                return;
            }
            if (callback)
                callback({
                    onlineNotes: Object.keys(notes).length,
                    onlineUsers: Object.keys(users).length,
                    distinctOnlineUsers: distinctaddresses.length,
                    notesCount: data.rows[0].count,
                    registeredUsers: regcount,
                    onlineRegisteredUsers: regusers,
                    distinctOnlineRegisteredUsers: distinctregusers
                });
        });
    });
}

function getNotenameFromSocket(socket) {
    var hostUrl = url.parse(socket.handshake.headers.referer);
    var notename = hostUrl.pathname.split('/')[1];
    if (notename == config.featuresnotename) {
        return notename;
    }
    if (!Note.checkNoteIdValid(notename)) {
        socket.emit('info', {
            code: 404
        });
        return socket.disconnect();
    }
    notename = LZString.decompressFromBase64(notename);
    return notename;
}

function emitOnlineUsers(socket) {
    var notename = getNotenameFromSocket(socket);
    if (!notename || !notes[notename]) return;
    var users = [];
    Object.keys(notes[notename].users).forEach(function (key) {
        var user = notes[notename].users[key];
        if (user)
            users.push(buildUserOutData(user));
    });
    notes[notename].socks.forEach(function (sock) {
        var out = {
            users: users
        };
        out = LZString.compressToUTF16(JSON.stringify(out));
        sock.emit('online users', out);
    });
}

function emitUserStatus(socket) {
    var notename = getNotenameFromSocket(socket);
    if (!notename || !notes[notename]) return;
    notes[notename].socks.forEach(function (sock) {
        if (sock != socket) {
            var out = buildUserOutData(users[socket.id]);
            sock.emit('user status', out);
        }
    });
}

var isConnectionBusy = false;
var connectionSocketQueue = [];
var isDisconnectBusy = false;
var disconnectSocketQueue = [];

function finishConnection(socket, notename) {
    notes[notename].users[socket.id] = users[socket.id];
    notes[notename].socks.push(socket);
    emitOnlineUsers(socket);
    socket.emit('refresh', {
        body: notes[notename].body
    });

    //clear finished socket in queue
    for (var i = 0; i < connectionSocketQueue.length; i++) {
        if (connectionSocketQueue[i].id == socket.id)
            connectionSocketQueue.splice(i, 1);
    }
    //seek for next socket
    isConnectionBusy = false;
    if (connectionSocketQueue.length > 0)
        startConnection(connectionSocketQueue[0]);

    if (config.debug) {
        console.log('SERVER connected a client to [' + notename + ']:');
        console.log(JSON.stringify(users[socket.id]));
        //console.log(notes);
        getStatus(function (data) {
            console.log(JSON.stringify(data));
        });
    }
}

function startConnection(socket) {
    if (isConnectionBusy) return;
    isConnectionBusy = true;

    var notename = getNotenameFromSocket(socket);
    if (!notename) return;

    if (!notes[notename]) {
        db.readFromDB(notename, function (err, data) {
            if (err) {
                socket.emit('info', {
                    code: 404
                });
                socket.disconnect();
                //clear err socket in queue
                for (var i = 0; i < connectionSocketQueue.length; i++) {
                    if (connectionSocketQueue[i].id == socket.id)
                        connectionSocketQueue.splice(i, 1);
                }
                isConnectionBusy = false;
                return console.error(err);
            }
            var body = LZString.decompressFromBase64(data.rows[0].content);
            body = LZString.compressToUTF16(body);
            notes[notename] = {
                socks: [],
                body: body,
                isDirty: false,
                users: {}
            };
            finishConnection(socket, notename);
        });
    } else {
        finishConnection(socket, notename);
    }
}

function disconnect(socket) {
    if (isDisconnectBusy) return;
    isDisconnectBusy = true;

    if (config.debug) {
        console.log("SERVER disconnected a client");
        console.log(JSON.stringify(users[socket.id]));
    }
    var notename = getNotenameFromSocket(socket);
    if (!notename) return;
    if (users[socket.id]) {
        delete users[socket.id];
    }
    if (notes[notename]) {
        delete notes[notename].users[socket.id];
        var index = notes[notename].socks.indexOf(socket);
        if (index > -1) {
            notes[notename].socks.splice(index, 1);
        }
        if (Object.keys(notes[notename].users).length <= 0) {
            var body = LZString.decompressFromUTF16(notes[notename].body);
            var title = Note.getNoteTitle(body);
            body = LZString.compressToBase64(body);
            db.saveToDB(notename, title, body,
                function (err, result) {
                    delete notes[notename];
                    if (config.debug) {
                        //console.log(notes);
                        getStatus(function (data) {
                            console.log(JSON.stringify(data));
                        });
                    }
                });
        }
    }
    emitOnlineUsers(socket);

    //clear finished socket in queue
    for (var i = 0; i < disconnectSocketQueue.length; i++) {
        if (disconnectSocketQueue[i].id == socket.id)
            disconnectSocketQueue.splice(i, 1);
    }
    //seek for next socket
    isDisconnectBusy = false;
    if (disconnectSocketQueue.length > 0)
        disconnect(disconnectSocketQueue[0]);

    if (config.debug) {
        //console.log(notes);
        getStatus(function (data) {
            console.log(JSON.stringify(data));
        });
    }
}

function buildUserOutData(user) {
    var out = {
        id: user.id,
        login: user.login,
        userid: user.userid,
        color: user.color,
        cursor: user.cursor,
        name: user.name,
        idle: user.idle,
        type: user.type
    };
    return out;
}

function updateUserData(socket, user) {
    //retrieve user data from passport
    if (socket.request.user && socket.request.user.logged_in) {
        var profile = JSON.parse(socket.request.user.profile);
        user.name = profile.displayName || profile.username;
        user.userid = socket.request.user._id;
        user.login = true;
    } else {
        user.userid = null;
        user.name = 'Guest ' + chance.last();
        user.login = false;
    }
}

function connection(socket) {
    //split notename from socket
    var notename = getNotenameFromSocket(socket);

    //initialize user data
    //random color
    var color = randomcolor({
        luminosity: 'light'
    });
    //make sure color not duplicated or reach max random count
    if (notename && notes[notename]) {
        var randomcount = 0;
        var maxrandomcount = 5;
        var found = false;
        do {
            Object.keys(notes[notename].users).forEach(function (user) {
                if (user.color == color) {
                    found = true;
                    return;
                }
            });
            if (found) {
                color = randomcolor({
                    luminosity: 'light'
                });
                randomcount++;
            }
        } while (found && randomcount < maxrandomcount);
    }
    //create user data
    users[socket.id] = {
        id: socket.id,
        address: socket.handshake.address,
        'user-agent': socket.handshake.headers['user-agent'],
        otk: shortId.generate(),
        color: color,
        cursor: null,
        login: false,
        userid: null,
        name: null,
        idle: false,
        type: null
    };
    updateUserData(socket, users[socket.id]);

    //start connection
    connectionSocketQueue.push(socket);
    startConnection(socket);

    //when a new client coming or received a client refresh request
    socket.on('refresh', function (body_) {
        var notename = getNotenameFromSocket(socket);
        if (!notename) return;
        if (config.debug)
            console.log('SERVER received [' + notename + '] data updated: ' + socket.id);
        if (notes[notename].body != body_) {
            notes[notename].body = body_;
            notes[notename].isDirty = true;
        }
    });

    //received user status
    socket.on('user status', function (data) {
        var notename = getNotenameFromSocket(socket);
        if (!notename) return;
        if (config.debug)
            console.log('SERVER received [' + notename + '] user status from [' + socket.id + ']: ' + JSON.stringify(data));
        if (data) {
            var user = users[socket.id];
            user.idle = data.idle;
            user.type = data.type;
        }
        emitUserStatus(socket);
    });

    //reveiced when user logout or changed
    socket.on('user changed', function () {
        console.log('user changed');
        var notename = getNotenameFromSocket(socket);
        if (!notename || !notes[notename]) return;
        updateUserData(socket, notes[notename].users[socket.id]);
        emitOnlineUsers(socket);
    });

    //received sync of online users request
    socket.on('online users', function () {
        var notename = getNotenameFromSocket(socket);
        if (!notename || !notes[notename]) return;
        var users = [];
        Object.keys(notes[notename].users).forEach(function (key) {
            var user = notes[notename].users[key];
            if (user)
                users.push(buildUserOutData(user));
        });
        var out = {
            users: users
        };
        out = LZString.compressToUTF16(JSON.stringify(out));
        socket.emit('online users', out);
    });

    //check version
    socket.on('version', function () {
        socket.emit('version', config.version);
    });

    //received cursor focus
    socket.on('cursor focus', function (data) {
        var notename = getNotenameFromSocket(socket);
        if (!notename || !notes[notename]) return;
        users[socket.id].cursor = data;
        var out = buildUserOutData(users[socket.id]);
        notes[notename].socks.forEach(function (sock) {
            if (sock != socket) {
                sock.emit('cursor focus', out);
            }
        });
    });

    //received cursor activity
    socket.on('cursor activity', function (data) {
        var notename = getNotenameFromSocket(socket);
        if (!notename || !notes[notename]) return;
        users[socket.id].cursor = data;
        var out = buildUserOutData(users[socket.id]);
        notes[notename].socks.forEach(function (sock) {
            if (sock != socket) {
                sock.emit('cursor activity', out);
            }
        });
    });

    //received cursor blur
    socket.on('cursor blur', function () {
        var notename = getNotenameFromSocket(socket);
        if (!notename || !notes[notename]) return;
        users[socket.id].cursor = null;
        var out = {
            id: socket.id
        };
        notes[notename].socks.forEach(function (sock) {
            if (sock != socket) {
                if (sock != socket) {
                    sock.emit('cursor blur', out);
                }
            }
        });
    });

    //when a new client disconnect
    socket.on('disconnect', function () {
        disconnectSocketQueue.push(socket);
        disconnect(socket);
    });

    //when received client change data request
    socket.on('change', function (op) {
        var notename = getNotenameFromSocket(socket);
        if (!notename) return;
        op = LZString.decompressFromUTF16(op);
        if (op)
            op = JSON.parse(op);
        if (config.debug)
            console.log('SERVER received [' + notename + '] data changed: ' + socket.id + ', op:' + JSON.stringify(op));
        switch (op.origin) {
        case '+input':
        case '+delete':
        case '+transpose':
        case 'paste':
        case 'cut':
        case 'undo':
        case 'redo':
        case 'drag':
        case '*compose':
        case 'case':
        case '+insertLine':
        case '+swapLine':
        case '+joinLines':
        case '+duplicateLine':
        case '+sortLines':
            notes[notename].socks.forEach(function (sock) {
                if (sock != socket) {
                    if (config.debug)
                        console.log('SERVER emit sync data out [' + notename + ']: ' + sock.id + ', op:' + JSON.stringify(op));
                    sock.emit('change', LZString.compressToUTF16(JSON.stringify(op)));
                }
            });
            break;
        default: 
            console.log('SERVER received uncaught [' + notename + '] data changed: ' + socket.id + ', op:' + JSON.stringify(op));
        }
    });
}

module.exports = realtime;