'use strict';
const crypto = require('crypto');
const fs = require('fs')
const rp = require('request-promise');
const fetch = require('node-fetch');
const winston = require('winston')
const constants = require('.././constants_local.json')
const secrets = require('./secrets.json')
const metrics = require('./metrics')
const sharedMeta = require('./shared_meta')
const util = require('util')
const { createLogger, format, transports } = winston;
const heap = require('heap')
const dgram = require('dgram');
const udpProxy = dgram.createSocket('udp4');
//const req_make = require('request')
//const https = require('https')
const axios = require('axios')
// const indexfile = require('./index')

let struct = require('jspack');
const { resourceLimits } = require('worker_threads');
struct = struct.jspack


let db = sharedMeta.db, // queue holding request to be dispatched
   resourceMap = sharedMeta.resourceMap, // map between resource_id and resource details like node_id, port, associated function etc
   functionToResource = sharedMeta.functionToResource, // a function to resource map. Each map contains a minheap of
                                                      // resources associated with the function
   functionBranchTree = sharedMeta.functionBranchTree, // Holds the function path's and related probability distribution
   timelineQueue = new Map(), // a temporary map holding request timestamps to be used for calulcating implicit chain invocation delays
   requestFlightQueue = sharedMeta.requestFlightQueue,
   idToFunchashmap = sharedMeta.idToFunchashmap,
   resource_to_cpu_util = sharedMeta.resource_to_cpu_util,
   node_to_resource_mapping = sharedMeta.node_to_resource_mapping

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

let implicitChainDB = sharedMeta.implicitChainDB
/**
 * Generates unique IDs of arbitrary length
 * @param {Length of the ID} length 
 */
function makeid(length) {
    var result           = '';
    var characters       = 'abcdefghijklmnopqrstuvwxyz0123456789';
    var charactersLength = characters.length;
    for ( var i = 0; i < length; i++ ) {
       result += characters.charAt(Math.floor(Math.random() * charactersLength));
    }
    return result;
 }


/**
 * generates the runtime executor after inserting the received function
 * TODO: make this asynchronous
 * @param {string Path from where to extract the function} functionPath 
 * @param {string Function Hash value} functionHash 
 */
function generateExecutor(functionPath, functionHash) {

   let input = fs.readFileSync(`./repository/worker_env/${constants.env}`)
   let functionFile = fs.readFileSync(functionPath + functionHash)
   let searchSize = "(resolve, reject) => {".length

   let insertIndex = input.indexOf("(resolve, reject) => {") + searchSize

   let output = input.slice(0, insertIndex) + functionFile + input.slice(insertIndex)
    
   let hash = crypto.createHash('md5').update(output).digest("hex");
   let func_id = parseInt(hash.slice(0,5),16)
   console.log(func_id);
   
   // fs.writeFileSync(functionPath + hash + ".js", output)
   fs.writeFileSync(functionPath + "function_" + func_id + ".js", output )
   return "function_"+func_id
   // return hash
}

/**
 * generates the runtime executor after inserting the received function
 * TODO: make this asynchronous
 * @param {string Path from where to extract the function} functionPath 
 * @param {string Function Hash value} functionHash 
 */
 function generateMicrocExecutor(functionPath, functionName, jsfunctionhash) {
   //creating function.c
//      let function_temp = fs.readFileSync(`./repository/worker_env/function_temp.c`)
//      let function_def = fs.readFileSync(functionPath + functionName)
//      let searchSize = "//ADD_FUNCTION".length
//      let fid = parseInt(jsfunctionhash.slice(0,5), 16)
//      let insertIndex = function_temp.indexOf("//ADD_FUNCTION") + searchSize
//      let function_name = "void function_"+ fid +"(PIF_PLUGIN_map_hdr_T *mapHdr)"
   
//      let full_function = function_temp.slice(0, insertIndex) +"\n"+ function_name + "{\n" +function_def +"\n}"+ function_temp.slice(insertIndex)
    
//    //   let hash = crypto.createHash('md5').update(full_function).digest("hex");
//    //   console.log(hash);
//      console.log(full_function);
//      fs.writeFileSync(functionPath +"offload/"+ jsfunctionhash + ".c", full_function)
  
//   //adding call to function when match with fid
//   return new Promise((resolve) => {
//       let main_function_temp = fs.readFileSync(functionPath +"offload/"+ "static_dispatch_function.c")
//       // let client_function = fs.readFileSync(functionPath + "offload/"+jsfunctionhash+".c")
//       searchSize = "//ADD_FUNCTION_EXTERNS".length
//       insertIndex = main_function_temp.indexOf("//ADD_FUNCTION_EXTERNS") + searchSize
//       let extern_name = "extern void function_"+fid +"(PIF_PLUGIN_map_hdr_T *mapHdr)"
//       let main_function = main_function_temp.slice(0, insertIndex) +"\n"+ extern_name+";\n"+ main_function_temp.slice(insertIndex)
//       console.log("MAIN FUNCTION : \n",main_function)
//       let hash = crypto.createHash('md5').update(full_function).digest("hex");
//       // console.log(hash);
//       searchSize = "//ADD_FUNCTION_CONDITION".length
//       insertIndex = main_function.indexOf("//ADD_FUNCTION_CONDITION") + searchSize
//       let inc_pkt_count = "function_packet_count["+fid+"-10000]++;"
//       let if_else_cond = "else if( fid == "+fid + " ) {\n "+inc_pkt_count +"\nfunction_"+fid+"(mapHdr);\n}"
//       let main_function_full = main_function.slice(0, insertIndex) +"\n"+ if_else_cond +"\n"+ main_function.slice(insertIndex)

//       console.log(main_function_full);
//       fs.writeFileSync(functionPath +"offload/"+ "static_dispatch_function.c", main_function_full)
      // return 'xyz';
      // return hash
//   });
}

 /**
  * Reverse proxy to take user requests and forward them to appropriate workers using a loadbalacer
  * @param {JSON} req the user request to be forwarded to the worker
  * @param {JSON} res Object to use to return the response to the user
  */
async function reverseProxy(req, res) {
   //console.log("request is: ",req.body)
   //console.log("response is: ",res)
   console.log("reverseProxy called !!!")
   res.reverse_ingress = Date.now()
   if (req.headers['x-chain-type'] !== 'explicit' && req.body.type === "tcp")
      branchChainPredictor(req)
   let runtime = req.body.runtime
   //console.log(runtime==="process")
   let id = req.params.id + runtime
   /**
    * Bypass deployment pipeline if resource available
    */
   console.log("functionToResource : ", functionToResource)
   let functionHeap = functionToResource.get(id)
   console.log("functionHeap : ", functionHeap)
   // loadbalancing by choosing worker with lowest load
   let forwardTo = functionHeap[0]
   let resource = resourceMap.get(forwardTo.resource_id)
   // logger.info(`Choosing resource ${JSON.stringify(forwardTo.resource_id)}` +
   //    "\n forwarding via reverse proxy to: " + JSON.stringify(resource));
   let url = `http://${resource.node_id}:${resource.port}/serverless/function/execute`

   // logger.info("Request received at reverseproxy. Forwarding to: " + url);
   // forwardTo.open_request_count += 1l,
   // TODO: stopping loadbalancer
   // heap.heapify(functionHeap, compare) // maintain loadbalancer by heapifying the Map
   res.lookup_time = Date.now()
   var options = {
      method: 'POST',
      uri: url,
      body: req.body,
      json: true // Automatically stringifies the body to JSON
   };
   console.log("option created from reverse proxy : ", options, "body type : ", req.body , req.body.type) // new code
   if (req.body.type === "tcp") {
      console.log("tcp request to reverseproxy")
      try {
        // await new Promise(resolve => setTimeout(resolve, 5000));
         let parsedBody = await rp(options)
         let serviceTime = Date.now() - res.timestamp

         res.json(parsedBody)
         forwardTo.open_request_count -= 1
         heap.heapify(functionHeap, compare_uti)
         let functionHash = req.params.id
         let functionData = functionBranchTree.get(functionHash)

         if (functionData && functionData.req_count % 5 == 0) {
            if (functionData.parent)
               viterbi(functionHash, functionData)
            else {
               let head = await fetch(implicitChainDB + functionHash, {
                  method: "head"
               })

               functionData._rev = head.headers.get("etag").substring(1, head.headers.get("etag").length - 1)
               functionData.branches = Array.from(functionData.branches.entries())
               let payload = {
                  method: 'put',
                  body: JSON.stringify(functionBranchTree.get(functionHash)),
                  headers: { 'Content-Type': 'application/json' }
               }

               fetchData(implicitChainDB + functionHash, payload)
                  .then((updateStatus) => {
                     console.log(updateStatus);
                     if (updateStatus.error === undefined)
                        functionData._rev = updateStatus.rev
                  })
               functionData.branches = new Map(functionData.branches)
            }
         }
         metrics.collectMetrics({ type: res.start, value: serviceTime, functionHash: req.params.id, runtime })
      }
      catch (err) {
         res.json(err.message).status(err.statusCode)
         forwardTo.open_request_count -= 1
         heap.heapify(functionHeap, compare) 
         logger.error("error" + err)
      }
   } else if (req.body.type === "udp") {
      console.log("udp request to reverseproxy")
      let request_id = Math.floor(Math.random() * 1000)
      req.body.request_id = request_id
      // res.request_id = request_id
      requestFlightQueue.set(request_id, res)
      let payload = req.body
      payload.request_id = request_id
      let data = payload.data
      res.data_set_time = Date.now()
      let packet = packPacket({
         chain_id: 0,
         exec_id: request_id,
         function_id: 0,
         data,
         function_count: 1
      })
      res.pack_time = Date.now()
      udpProxy.send(packet, 0, packet.length, resource.port, resource.node_id, function (err, bytes) {
         logger.info(`forwarded request via UDP, IP 192.168.2.5 Port ${resource.port}`)
         res.send_time = Date.now()
      })
   }
}



function getPort(usedPort) {
   let port = -1, ctr = 0
   do {
       let min = Math.ceil(30000);
       let max = Math.floor(60000);
       port = Math.floor(Math.random() * (max - min + 1)) + min;
       ctr += 1;
       if (ctr > 30000) {
           port = -1
           break
       }
   } while (usedPort.has(port))
   return port
}

const logger = winston.createLogger({
   level: 'info',
   format: winston.format.combine(
      format.timestamp(),
      format.json()
   ),
   defaultMeta: { module: 'Dispatch Manager' },
   transports: [
      //
      // - Write to all logs with level `info` and below to `combined.log` 
      // - Write all logs error (and below) to `error.log`.
      //
      new winston.transports.File({ filename: 'log/error.log', level: 'error' }),
      new winston.transports.File({ filename: 'log/combined.log' }),
      new winston.transports.Console({
         format: winston.format.combine(
            format.colorize({ all: true }),
            format.timestamp(),
            format.simple()
            
         )
      })
   ]
});

function compare(a, b) {
   return a.open_request_count - b.open_request_count
}

function compare_uti(a, b) {
   return a.cpu_utilization - b.cpu_utilization
}

async function branchChainPredictor(req) {
   // console.log(req.headers['x-resource-id']);
   let destinationTimestamp = Date.now()
   if (!functionBranchTree.has(req.params.id)) {
      let data = await fetchData(implicitChainDB + req.params.id)
      if (data.error !== "not_found") {
         data.branches = new Map(data.branches)
         functionBranchTree.set(req.params.id, data)
      }
   }

   if (functionBranchTree.has(req.params.id) && functionBranchTree.get(req.params.id).branches.size > 0) {
      // console.log(timelineQueue.has(req.params.id), timelineQueue.get(req.params.id));
      
      if (!timelineQueue.has(req.params.id)) {
         timelineQueue.set(req.params.id, [])
      }
      timelineQueue.get(req.params.id).push(destinationTimestamp)
   }

   if (req.headers['x-resource-id'] === undefined) {

      let functionHash = req.params.id
      if (functionBranchTree.has(functionHash)) {
         let branchInfo = functionBranchTree.get(functionHash)
         branchInfo.req_count++

      } else {
         
         let data = {
            req_count: 1,
            parent: true,
            branches: new Map()
         }
         functionBranchTree.set(functionHash, data)
      }
      
   } else {
      let resource_id = req.headers['x-resource-id']
      let resource = resourceMap.get(resource_id)
      let forwardBranch = req.params.id, callDelay = 0

      if (timelineQueue.has(resource.functionHash)) {
         let sourceTimestamp = timelineQueue.get(resource.functionHash).shift()
         callDelay = destinationTimestamp - sourceTimestamp
         // console.log("callDelay", callDelay);

      }

      if (!functionBranchTree.has(resource.functionHash)) {
         let data = {
            req_count: 1,
            parent: false,
            branches: new Map()
         }
         data.branches.set(forwardBranch, [1, callDelay])
         functionBranchTree.set(resource.functionHash, data)
      } else {
         let branchInfo = functionBranchTree.get(resource.functionHash)
         if (!branchInfo.parent)
            branchInfo.req_count++
         if (branchInfo.branches.has(forwardBranch)) {
            callDelay = constants.metrics.alpha * branchInfo.branches.get(forwardBranch)[1] 
               + callDelay * (1 - constants.metrics.alpha)
            console.log("call delay", callDelay);
            
            let branchProb = branchInfo.branches.get(forwardBranch)[0]
            branchProb = (branchProb * (branchInfo.req_count - 1) + 1.0)
            branchInfo.branches.set(forwardBranch, [branchProb, callDelay])
         } else {
            branchInfo.branches.set(forwardBranch, [1.0, callDelay])
         }
         for (let [branch, prob] of branchInfo.branches.entries()) {
            if (branch !== forwardBranch)
               prob[0] *= (branchInfo.req_count - 1)
            prob[0] /= branchInfo.req_count
         }
      }
   }
   // console.log("timelineQueue", timelineQueue);
   
   // console.log("branch tree", util.inspect(functionBranchTree, false, null, true /* enable colors */));
}

async function viterbi(node, metadata) {
   let path = []
   let parents = [[node, {
      prob: 1,
      metadata
   }]]
   path.push({ node, probability: 1 })
   let siblings = new Map()
   while (parents.length > 0) {
      // console.log("parent_group", parents);

      for (const parent of parents) {
         // console.log("=========begin==========\n",parent, "\n=============end============");
         // console.log(parent[1].metadata);

         if (parent[1].metadata === undefined)
            continue
         let forwardBranches = parent[1].metadata.branches
         // console.log(forwardBranches);

         let parentProbability = parent[1].prob

         forwardBranches.forEach((branchProb, subNode) => {
            let probability = 0
            if (siblings.has(subNode))
               probability = siblings.get(subNode)
            probability += branchProb[0] * parentProbability
            // console.log("prob", probability);

            siblings.set(subNode, probability)
         })
         // console.log("siblings", siblings);

      }
      parents = []
      let maxSibling, maxProb = 0
      siblings.forEach((prob, sibling) => {
         if (prob > maxProb) {
            maxSibling = sibling
            maxProb = prob
         }
      })
      let parentIDs = Array.from(siblings.keys());
      for (const id of parentIDs) {
         let metadata = functionBranchTree.get(id)
         parents.push([
            id, {
               prob: siblings.get(id),
               metadata
            }
         ])
      }
      if (maxSibling !== undefined)
         path.push({ node: maxSibling, probability: maxProb })
      siblings = new Map()
   }
   if (path.length > 1)
      console.log("path", path);
   
   if (path.length > 1) {
      metadata.mle_path = path
      metadata.branches = Array.from(metadata.branches.entries())
      let head = await fetch(implicitChainDB + node, {
         method: "head"
      })

      metadata._rev = head.headers.get("etag").substring(1, head.headers.get("etag").length - 1)
      
      let payload = {
         method: 'put',
         body: JSON.stringify(functionBranchTree.get(node)),
         headers: { 'Content-Type': 'application/json' }
      }

      fetchData(implicitChainDB + node, payload)
         .then((updateStatus) => {
            console.log(updateStatus);
            if (updateStatus.error === undefined)
               metadata._rev = updateStatus.rev
         })
      metadata.branches = new Map(metadata.branches)
   }
}

function logBroadcast(message, resource_id) {
   return new Promise((resolve, reject) => {
      try {

         message.timestamp = Date.now()
         if (resource_id && resourceMap.has(resource_id)) {
            let resource = resourceMap.get(resource_id)
            message.resource_id = resource_id
            message.node_id = resource.node_id
            message.runtime = resource.runtime
            message.function_id = resource.functionHash
         }
         let log = [{
            topic: constants.topics.log_channel,
            messages: JSON.stringify(message),
            partition: 0
         }]
         producer.send(log, () => {
            resolve()
         })
      } catch (err) {
         console.log(err);
         reject()
      }
   })
   
}

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

udpProxy.on('message', async (msg, rinfo) => {
   //console.log("request has come?")
   let result = unpackPacket(msg)
   console.log("udp received request !!", result)
   
   console.log("idtofucntion hash map : ", idToFunchashmap)
   let func_id = "function_" + result.function_id 
   // let funchash = idToFunchashmap.get(func_id)
   let funchash = func_id

   try{
   const res2 = await axios({
      method: 'post',
      url: 'http://10.129.2.182:8082/serverless/execute/' + funchash,
      headers: {}, 
      data: {
        runtime: 'container' // This is the body part
      },
    });

    console.log(res2)
   }
   catch(err)
   {
      console.error(err)
   }

   // indexfile.dispatch()
   // res.json(result)
   // console.log("resource_lookup",
   //    res.dispatch_time - res.timestamp,
   //    "reverse_proxy_call",
   //    res.reverse_ingress - res.dispatch_time,
   //    "metadata_lookup",
   //    res.lookup_time - res.reverse_ingress,
   //    "data_set_time",
   //    res.data_set_time - res.lookup_time,
   //    "pack_time",
   //    res.pack_time - res.data_set_time,
   //    "network_send",
   //    res.send_time - res.pack_time,
   //    "total_dispatch_delay", 
   //    res.send_time - res.timestamp,
   //    "E2E time:", 
   //    Date.now() - res.timestamp
   // )
});

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

async function fetchData(url, data = null) {
   let res
   if (data === undefined || data === null)
      res = await fetch(url)
   else
      res = await fetch(url, data)
   return await res.json()
}

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 += 4
   data = struct.Unpack(">I", packet, base)
   base += 4
   function_count = struct.Unpack("B", packet, base)


   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
   let f0, f1, f2, f3, f4, t1, t2, t3, t4
   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])
   base += 1

   f0 = struct.PackTo("B", message, base, [0])
   base += 1
   f1 = struct.PackTo("B", message, base, [12])
   base += 1
   f2 = struct.PackTo("B", message, base, [0])
   base += 1
   f3 = struct.PackTo("B", message, base, [34])
   base += 1
   f4 = struct.PackTo("B", message, base, [0])
   base += 1

   t1 = struct.PackTo("I", message, base, [Date.now()])
   base += 4
   t2 = struct.PackTo("I", message, base, [1234])
   base += 4
   t3 = struct.PackTo("I", message, base, [0])
   base += 8
   t4 = struct.PackTo("I", message, base, [0])
   
   message = Buffer.from(message)
   return message
}


udpProxy.bind(constants.master_port); // starting UDP server for offloaded endpoints

 module.exports = {
    makeid, generateExecutor, generateMicrocExecutor, reverseProxy, 
    getPort, logger, compare, compare_uti,
    logBroadcast, fetchData, metrics,
    producer
 }
