'use strict';
const constants = require(".././constants_local.json")
const secrets = require('./secrets.json')
const config = require('./config.json')
const libSupport = require('./lib')
libSupport.updateConfig()
const node_id = config.id
const {spawn } = require('child_process')
const execute = require('./execute')
const fs = require('fs')
const fetch = require('node-fetch');
const os = require('os');
const dgram = require('dgram');
const server = dgram.createSocket('udp4');

let struct = require('jspack');
struct = struct.jspack

let metadataDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
metadataDB = metadataDB + "/" + constants.db.function_meta + "/"

const kafka = require('kafka-node')
const logger = libSupport.logger

const local_repository = __dirname + "/local_repository/"
const host_url = "http://" + constants.master_address + ":" + constants.master_port

let Producer = kafka.Producer,
    client = new kafka.KafkaClient({ 
        kafkaHost: constants.network.external.kafka_host,
        autoConnect: true
    }),
    producer = new Producer(client),
    Consumer = kafka.Consumer

libSupport.makeTopic(node_id).then(() => {
    logger.info("node topic created")
    let consumer = new Consumer(client,
        [
            { topic: node_id, partition: 0, offset: 0 }
        ],
        [
            { autoCommit: true }
        ])
    consumer.on('message', function (message) {
        // logger.info(message);
        let topic = message.topic
        message = message.value
        message = JSON.parse(message)
        let runtime = message.runtime
        let functionHash = message.functionHash
        let resource_id = message.resource_id
        let port = message.port
        /**
         * Download necessary files (function file) and Start resource deployment
         */
        if (message.type === "execute" && topic === node_id) {
            logger.info("Received Deployment request for resource_id: " + resource_id);
            fetch(metadataDB + functionHash).then(res => res.json())
            .then(json => {
                console.log("metadata", json);
                
                libSupport.download(host_url + "/repository/" + functionHash + ".js", local_repository + functionHash + ".js").then(() => {
                    let metadata = {
                        resource_id, functionHash,
                        runtime, port,
                        resources: {
                            memory: json.memory
                        }
                    }
                    startWorker(local_repository, producer, metadata)
            })
            }).catch(err => {
                logger.error("something went wrong" + err.toString())
            });
            
        }

    })
})

/**
 * download and start grunt
 */
libSupport.download(constants.grunt_host, "grunt", false).then(() => {
    logger.info("Downloaded grunt binary from repository")
    fs.chmod('grunt', 0o755, (err) => {
        logger.info("grunt made executable. Starting grunt")
        let grunt = spawn('./grunt', [node_id])
        grunt.stdout.on('data', data => {
            // logger.info(data.toString());

        })

        grunt.stderr.on('data', data => {
            // logger.info(data.toString());

        })
        grunt.on('close', (code) => {
            logger.info("Grunt exited with exit code", code);

        })
    })

})

    
/**
 * Start a worker executor of the runtime type
 * @param {String} local_repository 
 * @param {String} functionHash 
 * @param {String} resource_id 
 * @param {String} producer 
 * @param {String} runtime 
 * @param {Number} port 
 */
function startWorker(local_repository, producer, metadata) {
    let runtime = metadata.runtime
    console.log(metadata);
    
    logger.info(`Using port ${metadata.port} for functionHash ${metadata.functionHash}`)
    
    if (runtime === "isolate")
        execute.runIsolate(local_repository, metadata)
        .catch(err => {
            logger.error("=====================deployment failed=========================");
            logger.error(err)
            producer.send([{
                topic: "deployed",
                messages: JSON.stringify({
                    "status": false,
                    resource_id: metadata.resource_id,
                    "reason": "isolate exit"
                })
            }], () => { })
        })
    else if (runtime === "process")
//	console.log("rutime is process : ",metadata)
        execute.runProcess(local_repository, metadata)
        .catch(err => {
            logger.error("=====================deployment failed=========================");
            producer.send([{ topic: "deployed",
                messages: JSON.stringify({ 
                "status": false, 
                resource_id: metadata.resource_id,
                "reason": "process exit"
            }) }], () => { })
        })
    else if (runtime === "container")
    {
        console.log("rutime is container : ",metadata)
        execute.runContainer(metadata)
    }
    else {
        producer.send(
            [{
                topic: "response",
                messages: JSON.stringify({ status: "unknown runtime" })
            }], () => { })

        return
    }
    
}

function heartbeat() {
    let info = {
        free_mem: os.freemem(),
        cpu_count: os.cpus().length,
        total_mem: os.totalmem(),
        avg_load: os.loadavg()
    }
    let payload = [{
        topic: "heartbeat",
        messages: JSON.stringify({
            "address": node_id,
            "port": constants.daemon_port,
            "mac": constants.daemon_mac,
            "system_info": info,
            "timestamp": Date.now()
        })
    }]
    // console.log("daemon system info : ", info)
    producer.send(payload, function(cb) {})
}

// TODO 2: implement packer deparser for the udp packet
// TODO 3: create UPD server to get the coldstart request 
server.on('error', (err) => {
    console.log(`server error:\n${err.stack}`);
    server.close();
});


server.on('message', (msg, rinfo) => {
    console.log("message", msg)
    let payload = unpackPacket(msg)
    console.log(payload, typeof payload);
    // get the coldstart request and start the function
    // logger.info("Received Deployment UDP request for resource_id: " + resource_id);
    let functionHash = "function_" + payload.function_id
    let resource_id = 'aaa'
    let runtime = 'process'
    let port = 9920
    let mac = constants.daemon_mac
    logger.info("Received Deployment UPD request")
    fetch(metadataDB + functionHash).then(res => res.json())
        .then(json => {
            console.log("metadata", json);
            
            libSupport.download(host_url + "/repository/" + functionHash + ".js", local_repository + functionHash + ".js").then(() => {
                let metadata = {
                    resource_id, functionHash,
                    runtime, port, mac,
                    resources: {
                        memory: json.memory
                    }
                }
                
                startWorker(local_repository, producer, metadata)
        })
        }).catch(err => {
            logger.error("something went wrong" + err.toString())
        });


    // 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("I", packet, base)
    base += 4

    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(constants.daemon_port);

setInterval(heartbeat, 1000);
