// LICENSE_CODE ZON
'use strict'; /*zlint node*/
require('./config.js');
const zerr = require('./zerr.js');
const etask = require('./etask.js');
const zcounter = require('./zcounter.js');
const zutil = require('./util.js');
const date = require('./date.js');
const array = require('./array.js');
const slow = require('./slow.js');
const events = require('events');
const http = require('http');
const util = require('util');
const ws = require('ws');
let uws;
if (!/^win/.test(process.platform) && !process.env.NODE_EMBEDDED)
    uws = require('uws');
const E = exports, env = process.env, ms = date.ms, ef = etask.ef;
const assign = Object.assign;

E.send = (conn, msg, opt)=>etask(function*ws_send(){
    opt = opt||{};
    if (!conn || conn.readyState!=1)
        throw new Error('ws.send: not connected');
    let json = msg instanceof Buffer ? msg : JSON.stringify(msg);
    if (!opt.no_timeout)
        this.alarm(opt.timeout||30*date.ms.SEC, {throw: 'timeout'});
    try {
        return yield etask.nfn_apply({cancel: true}, conn, '.send', [json]);
    } catch(e){ ef(e);
        zerr('ws_send_err: %O %O', zerr.e2s(e), msg instanceof Buffer ?
            msg.slice(0, 128) : json.substr(0, 1024));
        zcounter.inc('ws_send_'+(e=='timeout' ? 'timeout' : 'err'));
        if (conn)
            conn.close();
    }
});

const ipc_handler_call = (conn, msg)=>etask(function*ws_ipc_handler_call(){
    if (!!+env.WS_IPC_LOG)
    {
        zerr.notice('ws_call: '+msg.cmd+' from '+conn._socket.remoteAddress+
            ': %O', msg.msg);
    }
    let data = {cmd: msg.cmd, cookie: msg.cookie, type: 'ipc_result'};
    try {
        if (!msg.cmd)
            throw new Error('no_cmd');
        if (!conn.ipc_handlers[msg.cmd])
            throw new Error('no_handler');
        data.msg = yield conn.ipc_handlers[msg.cmd](msg.msg, conn);
    } catch(e){ ef(e);
        data.type = 'ipc_error';
        data.msg = zerr.e2s(e);
    }
    return yield E.send(conn, data);
});

const ipc_handler_call_sync = (conn, msg)=>{
    if (!!+env.WS_IPC_LOG)
    {
        zerr.notice('ws_call: '+msg.cmd+' from '+conn._socket.remoteAddress+
            ': %O', msg.msg);
    }
    let data = {cmd: msg.cmd, cookie: msg.cookie, type: 'ipc_result'};
    try {
        if (!msg.cmd)
            throw new Error('no_cmd');
        if (!conn.ipc_handlers[msg.cmd])
            throw new Error('no_handler');
        data.msg = conn.ipc_handlers[msg.cmd](msg.msg, conn);
    } catch(e){ ef(e);
        data.type = 'ipc_error';
        data.msg = zerr.e2s(e);
    }
    return E.send(conn, data);
};

function ipc_handler_resp(conn, msg){
    let key = msg.cmd+'-'+msg.cookie;
    let handler = zutil.obj_pluck(conn.ipc_waiting, key);
    if (!handler)
        return zerr('ws_ipc got response for non-exists call: %O', msg);
    if (msg.type=='ipc_result')
        handler.continue(msg.msg);
    if (msg.type=='ipc_error')
        handler.throw(msg.msg);
}

E.ipc_init = (conn, opt)=>{
    if (conn.ipc_cookie)
        throw new Error('ws.ipc_init already initialized');
    conn.ipc_handlers = {};
    conn.ipc_waiting = {};
    conn.ipc_cookie = 1;
    conn.ev.on('parsed_message', (data, msg)=>{
        try {
            if (!data.cookie || !data.type || !data.cmd)
                return;
            if (data.type=='ipc_call')
            {
                if (opt.ipc_sync)
                    ipc_handler_call_sync(conn, data);
                else
                    ipc_handler_call(conn, data);
            }
            if (data.type=='ipc_result' || data.type=='ipc_error')
                ipc_handler_resp(conn, data);
        }
        catch(e){ zerr('ws_ipc: '+e+' '+msg); }
    });
    conn.ev.on('close', ()=>{
        let keys = Object.keys(conn.ipc_waiting);
        for (let i=0; i<keys.length; i++)
            conn.ipc_waiting[keys[i]].throw('ws_sock_err');
    });
};

E.ipc_handler = (conn, cmd, handler)=>{
    if (conn.ipc_handlers[cmd])
        throw new Error('ws.ipc_handler already installed: '+cmd);
    conn.ipc_handlers[cmd] = handler;
};

E.ipc_call = (conn, cmd, msg, opt)=>etask(function*(){
    let retry = opt&&opt.retry||1, res;
    let retry_period = opt&&opt.retry_period||10*date.ms.SEC;
    let _slow = !opt || opt.slow==null ? true : opt.slow;
    for (let attempt=0; attempt<retry; attempt++)
    {
        try {
            return yield E._ipc_call(conn, cmd, msg, opt&&opt.timeout, _slow);
        } catch(e){ ef(e);
            if (attempt==retry-1)
            {
                zcounter.inc('ws_ipc_call_err');
                throw e;
            }
            zerr.notice(`ws.ipc_call, cmd: ${cmd} (attempt ${attempt+1}) `+
                `error${e.message?': '+e.message:''}`);
            yield etask.sleep(retry_period);
        }
    }
    return res;
});

const slow_ipc_opt = slow.slow_opt('ws_ipc_call',
    (start_time, et)=>'ws.ipc_call '+et.info.cmd);

E._ipc_call = (conn, cmd, msg, timeout, _slow)=>etask(function*ws_ipc_call(){
    if (!conn)
        throw new Error('ws.ipc_call not connected');
    if (!cmd)
        throw new Error('ws.ipc_call no cmd: '+msg);
    this.info.cmd = cmd;
    if (_slow)
        slow.etask(this, slow_ipc_opt);
    let cookie = conn.ipc_cookie++;
    let key = cmd+'-'+cookie;
    let w = conn.ipc_waiting[key] = etask.wait();
    this.finally(()=>delete conn.ipc_waiting[key]);
    let _timeout = timeout||conn.timeout||30*ms.SEC;
    this.alarm(_timeout, {throw: 'timeout'});
    yield E.send(conn, {cookie: cookie, cmd: cmd, msg: msg,
        type: 'ipc_call'});
    return yield w;
});

E.msg_call = (conn, cmd, data)=>{
    return E.send(conn, {cmd: cmd, msg: data}); };

const msg_handlers = (handlers, msg, conn)=>etask(function*msg_handlers(){
    try {
        for (let i = 0; i<handlers.length; i++)
            yield handlers[i](msg, conn);
    } catch(e){ ef(e);
        zerr('ws_msg_err: %s %O', e, msg);
        zcounter.inc('ws_msg_err');
    }
});

const msg_handlers_sync = (handlers, msg, conn)=>{
    try {
        for (let i=0; i < handlers.length; i++)
            handlers[i](msg, conn);
    } catch(e){ ef(e);
        zerr('ws_msg_err: %s %O', e, msg);
        zcounter.inc('ws_msg_err');
    }
};

E.msg_init = (conn, opt)=>{
    conn.msg_handlers = {};
    conn.ev.on('parsed_message', (data, msg)=>{
        try {
            let handlers = conn.msg_handlers[data.cmd];
            let d = data.msg||data.data||data.opt;
            if (!data.cmd||!d||!handlers||!handlers.length)
                return;
            if (opt.msg_sync)
                msg_handlers_sync(handlers, d, conn);
            else
                msg_handlers(handlers, d, conn);
        }
        catch(e){ zerr('ws_msg: '+e+' '+msg); }
    });
};

E.msg_parse = (conn, opt)=>{
    conn.on('message', (msg, flags)=>{
        if (flags&&flags.binary)
            return;
        let data;
        try { data = JSON.parse(msg); }
        catch(e){ return void zerr('ws.parse fail '+msg); }
        conn.ev.emit('parsed_message', data, msg);
    });
};

E.msg_handler = (conn, cmd, handler)=>{
    if (!conn || !cmd || !handler)
        throw new Error('ws.msg wrong args');
    if (!Array.isArray(conn.msg_handlers))
        conn.msg_handlers[cmd] = [];
    conn.msg_handlers[cmd].push(handler);
};

const ws_mod = opt=>opt.uws ? uws : ws;

E.server_init = (port, handler, opt)=>{
    opt = assign({ip_conn: true}, opt);
    let ws_server = new (ws_mod(opt).Server)(assign({perMessageDeflate: false},
        opt.ws_opt||{port: port}));
    ws_server.ip_conn = {};
    ws_server.on('connection', conn=>{
        let ip = conn.upgradeReq.headers['x-forwarded-for']
            || conn._socket.remoteAddress;
        // uws does not allow more than one listener - use separate events
        conn.ev = opt.uws ? new events.EventEmitter() : conn;
        zerr.notice('ws.server ipc connection from: '+ip);
        if (opt.ip_conn)
        {
            if (!Array.isArray(ws_server.ip_conn[ip]))
                ws_server.ip_conn[ip] = [];
            ws_server.ip_conn[ip].push(conn);
        }
        zcounter.inc_level('ws_conn', 1, 'sum');
        if (opt.timeout && !opt.uws)
        {
            let timeout = opt.timeout===true ? ms.HOUR : opt.timeout;
            conn._socket.setTimeout(timeout, ()=>{
                zcounter.inc('ws_server_sock_timeout');
                zerr('ws_server_sock_timeout');
                if (conn)
                    conn.close();
            });
        }
        conn.on('error', err=>{
            zcounter.inc('ws_server_sock_err');
            zerr('ws_server_sock_err: '+err);
            if (conn)
                conn.close();
        });
        conn.on('close', ()=>{
            if (conn._socket && !opt.uws)
                conn._socket.destroy();
            if (opt.ip_conn)
                array.rm_elm(ws_server.ip_conn[ip], conn);
            if (opt.uws)
                conn.ev.emit('close');
            zcounter.inc_level('ws_conn', -1, 'sum');
        });
        if (!opt.no_ipc)
            E.ipc_init(conn, opt);
        if (!opt.no_msg)
            E.msg_init(conn, opt);
        if (!opt.no_ipc || !opt.no_msg || opt.parse_json)
            E.msg_parse(conn, opt);
        handler(conn);
    });
    ws_server.on('error', err=>zerr('ws_server: '+err));
    return ws_server;
};

E.connect = (url, handler, opt)=>etask(function*ws_auth(){
    this.on('uncaught', e=>zerr.notice('ws.connect error', e));
    opt = opt||{};
    let conn;
    this.finally(()=>{
        if (conn)
            conn.close();
    });
    for (;;)
    {
        let w;
        if (opt.url_cb)
            url = yield opt.url_cb();
        conn = new (ws_mod(opt))(url, opt.ws_opt);
        conn.ev = opt.uws ? new events.EventEmitter() : conn;
        if (opt.uws)
            conn.url = url;
        if (opt.conn_cb)
            opt.conn_cb(conn);
        let closed;
        conn.on('close', (code, message)=>{
            if (opt.uws&&conn)
                conn.ev.emit('close', code, message);
            if (closed)
                return;
            closed = true;
            zerr.notice('ws.connect closed');
            w.continue();
        });
        conn.on('error', err=>{
            if (opt.uws&&conn)
                conn.ev.emit('error', err);
            if (closed)
                return;
            closed = true;
            zerr('ws.connect: '+err);
            w.continue();
        });
        conn.on('open', ()=>{
            if (opt.uws&&conn)
                conn.ev.emit('open');
            zerr.notice('ws.connect connected to: '+url);
            if (!opt.no_ipc)
                E.ipc_init(conn, opt);
            if (!opt.no_msg)
                E.msg_init(conn, opt);
            if (!opt.no_ipc || !opt.no_msg || opt.parse_json)
                E.msg_parse(conn, opt);
            handler(conn);
        });
        yield w = this.wait();
        if (conn)
            conn = void conn.close();
        if (opt.no_retry)
            return;
        let sleep = opt.retry_cb ? opt.retry_cb() : 5*ms.SEC;
        yield etask.sleep(sleep);
    }
});

E.fast_server_init = opt=>{
    let ws_srv = new ws.Server({noServer: true});
    let srv = opt.srv||http.createServer((req, res)=>{
        var body = http.STATUS_CODES[404];
        res.writeHead(404, {'Content-Length': body.length,
            'Content-Type': 'text/plain'});
        res.end(body);
    });
    srv.on('upgrade', (req, socket, upgrade_head)=>{
        // copy upgrade_head to avoid retention of large slab buffers used in
        // node core
        let head = new Buffer(upgrade_head.length);
        upgrade_head.copy(head);
        if (req.headers['x-hola-ws-prot']==1)
        {
            let client = new ws([null, socket, head]);
            ws_srv.emit('connection', client);
        }
        else
        {
            ws_srv.handleUpgrade(req, socket, head, client=>{
                ws_srv.emit('connection', client); });
        }
    });
    if (!opt.srv)
        srv.listen(opt.port, ()=>zerr.notice('fast ws server listening'));
    return {ws_srv, srv};
};

E.fast_client_init = socket=>{
    let headers = ['GET / HTTP/1.1',
        'Upgrade: websocket',
        'Connection: Upgrade',
        'X-Hola-Ws-Prot: 1'];
    socket.write(headers.concat('', '').join('\r\n'));
    let ws_client = new ws([null, socket, null]);
    return ws_client;
};
