'use strict';
const express = require('express')
// let request = require('request')
const process = require('process')

const dgram = require('dgram');
const { constants } = require('buffer');
const { spawnSync, execSync } = require('child_process');
// const { logger } = require('../../lib');
const server = dgram.createSocket('udp4');
const udpProxy = dgram.createSocket('udp4');
// var usage = require('usage');
const os = require('os')
let struct = require('jspack')
struct = struct.jspack


const app = express()
let port = 5000, resource_id, functionHash, runtime, idleTime = 1000, flagFirstRequest = true
let waitTime

resource_id = process.argv[2]
functionHash = process.argv[3]
port = process.argv[4]
runtime = process.argv[5]
// request = request.defaults({
//     headers: { 'x-resource-id': resource_id }
// });

let producer
try {
    let kafka = require('kafka-node'),
        Producer = kafka.Producer,
        client = new kafka.KafkaClient({
            kafkaHost: process.argv[6],
            autoConnect: true
        })
    producer = new Producer(client)
} catch(e) {
    console.log("Exception: ", e);
}

// var startUsage = process.cpuUsage()
// var startDate = Date.now()
// function usage_metrics2() {
//     var usage = process.cpuUsage(startUsage)
//     var result = 100 * (usage.user + usage.system) / ((Date.now() - startDate) * 1000) 
//     startDate = Date.now()
//     startUsage = usage
//     console.log(result)
//     if (producer)
//         producer.send(
//         [{
//             topic: "metrics_worker",
//             messages: JSON.stringify({ functionHash, portExternal: port,
//                 runtime, resource_id, cpu_usage: result})
//         }], () => { })
// }
// function usage_metrics() {
// var options = { keepHistory: true }
// var pid = process.pid // you can use any valid PID instead
// usage.lookup(pid, options, function(err, result) {
//  console.log(result)
// });
//usage.clearHistory(pid); //clear history for the given pid

//}

//setInterval(usage_metrics, 1000);
//setInterval(usage_metrics2, 1000);
//let repository = __dirname + "../../dispatch_manager/repository/worker_env/"
//let filename = repository +  "metrics_template.js"

app.use(express.json());
app.use(express.urlencoded({ extended: true }));

//const ls = spawn("node", [filename, process.pid, process.argv[6]]);

let lastRequest = Date.now(), totalRequest = 0
app.post('/serverless/function/execute/', (req, res) => {
    if (flagFirstRequest) {
        waitTime = Date.now() - waitTime
        flagFirstRequest = false
    }
    let payload = req.body
    lastRequest = Date.now()
    totalRequest++
    executor(payload).then((result) => {
        res.json(result)
    })
})

app.post('/serverless/function/timeout', (req, res) => {
    console.log(req.body);
    
    let idleTime = req.body.timeout
    console.log("Idle time set to: ", idleTime);
    res.json({
        status: "success"
    })
})



async function executor(payload) {
    return new Promise((resolve, reject) => {
        
    })
}

app.listen(port, () => {
    console.log(`Resource ${resource_id} Server listening on port ${port}!`)
    let node_id, mac_address;
    let interfaces = os.networkInterfaces()
    for (let networkInterface in interfaces) {
        networkInterface = interfaces[networkInterface]
        if (networkInterface[0].address.startsWith("192.168.")) {
            node_id = networkInterface[0].address
            mac_address = networkInterface[0].mac
        }
    }
    console.log({
        topic: "deployed",
        messages: JSON.stringify({
            functionHash, portExternal: port, node_id: node_id.trim(),
            runtime, resource_id, entity_id: process.pid, mac: mac_address
        }),
        "status": true
    });
    if (producer)
        producer.send(
        [{
            topic: "deployed",
            messages: JSON.stringify({ functionHash, portExternal: port,  node_id,
                runtime, resource_id, entity_id: process.pid, mac: mac_address}),
            "status": true
        }], () => { })
    waitTime = Date.now()
})

function shouldDie() {
    
    if (Date.now() - lastRequest > idleTime * 100) {
        let message = JSON.stringify({
            functionHash, portExternal: port,
            runtime, resource_id, entity_id: process.pid,
            total_request: totalRequest, wait_time: waitTime
        })

        console.log("Idle for too long. Exiting");
        if (producer)
            producer.send(
                [
                    {topic: "removeWorker", messages: message }
                ], (err, data) => {
                    if (err)
                        console.log(err);
                        
                    console.log("Ending worker for function", functionHash, "resource_id", resource_id);
                    process.exit(0)
                })
        
    }
}


server.on('error', (err) => {
    console.log(`server error:\n${err.stack}`);
    server.close();
});

function getRandomInt(min, max) {
    min = Math.ceil(min);
    max = Math.floor(max);
    return Math.floor(Math.random() * (max - min + 1)) + min;
}

server.on('message', (msg, rinfo) => {
    console.log("message", msg)
    let payload = unpackPacket(msg)
    console.log(payload, typeof payload);
    lastRequest = Date.now()
    console.log("network stack time", lastRequest - payload.t1)
    totalRequest++
    executor(msg).then(result => {
        result = packPacket(msg)
        let port = 10000 + getRandomInt(0, 10)
        try {
            udpProxy.send(msg, 0, msg.length, port, rinfo.address, function (err, bytes) {
                if (err)
                    console.log(err)
                console.log("response via UDP")
            })
        } catch (e) {
            console.log(e)
        }

    })
});

function unpackPacket(packet) {
    // let buffer = new Array(1024)
    let chain_id = null, exec_id = null, function_count = null, function_id = null, data = null
    let base = 0, f0, f1, f2, f3, f4, t1, t2, t3, t4
    chain_id = struct.Unpack(">I", packet, base)
    base += 4
    exec_id = struct.Unpack(">I", packet, base)
    base += 4
    function_id = struct.Unpack(">I", packet, base)
    base += 4
    data = struct.Unpack(">I", packet, base)
    base += 4
    function_count = struct.Unpack("B", packet, base)
    base += 1

    f0 = struct.Unpack("B", packet, base)
    base += 1
    f1 = struct.Unpack("B", packet, base)
    base += 1
    f2 = struct.Unpack("B", packet, base)
    base += 1
    f3 = struct.Unpack("B", packet, base)
    base += 1
    f4 = struct.Unpack("B", packet, base)
    base += 1

    t1 = struct.Unpack("I", packet, base)
    base += 8
    t2 = struct.Unpack("I", packet, base)
    base += 8
    t3 = struct.Unpack("I", packet, base)
    base += 8
    t4 = struct.Unpack("I", packet, base)


    console.log("chain_id", chain_id, "exec_id", exec_id, "data", data, "function_count", function_count, "function_id", function_id)


    return {
        chain_id: chain_id[0],
        exec_id: exec_id[0],
        data: data[0],
        function_count: function_count[0],
        function_id: function_id[0],
        f0, f1, f2, f3, f4, t1, t2, t3, t4
    }
}

function packPacket(dataPacket) {

    let message = new Array(1024)
    let base = 0, chain_id, exec_id, function_id, data, function_count
    chain_id = struct.PackTo(">I", message, base, [dataPacket.chain_id])
    base += 4
    exec_id = struct.PackTo(">I", message, base, [dataPacket.exec_id])
    base += 4
    function_id = struct.PackTo(">I", message, base, [dataPacket.function_id])
    base += 4
    data = struct.PackTo(">I", message, base, [dataPacket.data])
    base += 4
    function_count = struct.PackTo("B", message, base, [dataPacket.function_count])
    message = Buffer.from(message)
    return message
}


server.on('listening', () => {
    const address = server.address();
    console.log(`server listening ${address.address}:${address.port}`);
});

// server.bind(port, "192.168.2.3");
server.bind(port);
//var fs = require('fs');

// var getUsage = function(cb){
//     fs.readFile("/proc/" + process.pid + "/stat", function(err, data){
//         var elems = data.toString().split(' ');
//         var utime = parseInt(elems[13]);
//         var stime = parseInt(elems[14]);

//         cb(utime + stime);
//     });
// }

// setInterval(function(){
//     getUsage(function(startTime){
//         setTimeout(function(){
//             getUsage(function(endTime){
//                 var delta = endTime - startTime;
//                 var percentage = 100 * (delta / 1000);

                 
//                 console.log("CPU Usage : ", percentage, "%!");
                
//             });
//         }, 1000);
//     });
// }, 1000);



//setInterval(usage_metrics, 1000);
setInterval(shouldDie, 1000);
//setInterval(usage_metrics2, 1000);
