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.json')
const secrets = require('./secrets.json')
const metrics = require('./metrics')
const sharedMeta = require('./shared_meta')
const { createLogger, format, transports } = winston;
const heap = require('heap')


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
   workerNodes = sharedMeta.workerNodes, // list of worker nodes currently known to the DM
   functionBranchTree = sharedMeta.functionBranchTree // Holds the function path's and related probability distribution

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) {
   input = fs.readFileSync('./repository/worker_env/env.js')
   functionFile = fs.readFileSync(functionPath + functionHash)
   searchSize = "(resolve, reject) => {".length

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

   output = input.slice(0, insertIndex) + functionFile + input.slice(insertIndex)
    
   let hash = crypto.createHash('md5').update(output).digest("hex");
   console.log(hash);
   
    fs.writeFileSync(functionPath + hash + ".js", output)
    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
  */
function reverseProxy(req, res) {
   if (req.headers['x-chain-type'] !== 'explicit')
      branchChainPredictor(req)
   let runtime = req.body.runtime
   let id = req.params.id + runtime
   /**
    * Bypass deployment pipeline if resource available
    */
   let functionHeap = functionToResource.get(id)
   // 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 += 1
   heap.heapify(functionHeap, compare) // maintain loadbalancer by heapifying the Map
   // logger.info(functionHeap);
   
   var options = {
      method: 'POST',
      uri: url,
      body: req.body,
      json: true // Automatically stringifies the body to JSON
   };
   
   
   rp(options)
      .then(function (parsedBody) {
         let serviceTime = Date.now() - res.timestamp
         
         res.json(parsedBody)
         forwardTo.open_request_count -= 1
         heap.heapify(functionHeap, compare)
         let functionHash = req.params.id
         let functionData = functionBranchTree.get(functionHash)
         if (functionData && functionData.req_count % 5 == 0) {
            if (functionData.parent)
               viterbi(functionHash, functionData)
            else {
               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(function (err) {
         forwardTo.open_request_count -= 1
         heap.heapify(functionHeap, compare)
         logger.error("error" + err);
         res.json(err.message).status(err.statusCode)
      });
}

function getPort(usedPort) {
   let port = -1, ctr = 0
   do {
       min = Math.ceil(30000);
       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
}

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

   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
      if (!functionBranchTree.has(resource.functionHash)) {
         let data = {
            req_count: 1,
            parent: false,
            branches: new Map()
         }
         data.branches.set(forwardBranch, 1)
         functionBranchTree.set(resource.functionHash, data)
      } else {
         let branchInfo = functionBranchTree.get(resource.functionHash)
         if (!branchInfo.parent)
            branchInfo.req_count++
         if (branchInfo.branches.has(forwardBranch)) {
            let branchProb = branchInfo.branches.get(forwardBranch)
            branchProb = (branchProb * (branchInfo.req_count - 1) + 1.0)
            branchInfo.branches.set(forwardBranch, branchProb)
         } else {
            branchInfo.branches.set(forwardBranch, 1.0)
         }
         for (let [branch, prob] of branchInfo.branches.entries()) {
            if (branch !== forwardBranch)
               prob *= (branchInfo.req_count - 1)
            prob /= branchInfo.req_count
            branchInfo.branches.set(branch, prob)
         }
      }
   }
   
   // console.log("branch tree", functionBranchTree);
}

async function viterbi(node, metadata) {
   console.log("function branch tree", functionBranchTree.get(node));
   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 * 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
         }
      })
      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);
   
   
   metadata.mle_path = path
   if (path.length > 1) {
      metadata.branches = Array.from(metadata.branches.entries())
      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()
      }
   })
   
}

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()
}

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