'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');
let struct = require('jspack')
struct = struct.jspack


const app = express()
let port = 5000, resource_id, functionHash, runtime, idleTime = 600, 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);
}


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

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 = execSync('ip a | grep -oE "([0-9]{1,3}\.){3}[0-9]{1,3}" | grep 192.168 | head -n 1').toString()
    console.log({
        topic: "deployed",
        messages: JSON.stringify({
            functionHash, portExternal: port, node_id: node_id.trim(),
            runtime, resource_id, entity_id: process.pid
        }),
        "status": true
    });
    if (producer)
        producer.send(
        [{
            topic: "deployed",
            messages: JSON.stringify({ functionHash, portExternal: port,  node_id,
                runtime, resource_id, entity_id: process.pid}),
            "status": true
        }], () => { })
    waitTime = Date.now()
})

function shouldDie() {
    
    if (Date.now() - lastRequest > idleTime * 1000) {
        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();
});

server.on('message', (msg, rinfo) => {
    // console.log("message", msg)
    let payload = unpackPacket(msg)
    console.log(payload, typeof payload);
    lastRequest = Date.now()
    totalRequest++
    executor(payload).then(result => {
        result = packPacket(payload)
        console.log(result)
        try {
            udpProxy.send(result, 0, result.length, "8080", "192.168.2.2", 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
    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 += 1
    data = struct.Unpack(">I", packet, base)
    base += 4
    function_count = struct.Unpack("B", 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]
    }
}

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 += 1
    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);

setInterval(shouldDie, 1000);
