'use strict'

const http = require('node:http')
const os = require('node:os')
const path = require('node:path')
const { Writable } = require('node:stream')
const { isMainThread } = require('node:worker_threads')

const { Pool, Client, fetch, Agent, setGlobalDispatcher } = require('..')

const { makeParallelRequests, printResults } = require('./_util')

const { cronometro } = require('cronometro')
const { default: nodeFetch } = require('node-fetch')
const axios = require('axios')
const _superagent = require('superagent')
const { default: got } = require('got')

const { promisify } = require('node:util')
const request = promisify(require('request'))

const iterations = (parseInt(process.env.SAMPLES, 10) || 10) + 1
const parsedErrorThreshold = parseInt(process.env.ERROR_THRESHOLD, 10)
const errorThreshold = Number.isNaN(parsedErrorThreshold) ? 3 : parsedErrorThreshold
const connections = parseInt(process.env.CONNECTIONS, 10) || 50
const pipelining = parseInt(process.env.PIPELINING, 10) || 10
const headersTimeout = parseInt(process.env.HEADERS_TIMEOUT, 10) || 0
const bodyTimeout = parseInt(process.env.BODY_TIMEOUT, 10) || 0
const dest = {}

if (process.env.PORT) {
  dest.port = process.env.PORT
  dest.url = `http://localhost:${process.env.PORT}`
} else {
  dest.url = 'http://localhost'
  dest.socketPath = path.join(os.tmpdir(), 'undici.sock')
}

/** @type {http.RequestOptions} */
const httpBaseOptions = {
  protocol: 'http:',
  hostname: 'localhost',
  method: 'GET',
  path: '/',
  ...dest
}

/** @type {http.RequestOptions} */
const httpNoKeepAliveOptions = {
  ...httpBaseOptions,
  agent: new http.Agent({
    keepAlive: false,
    maxSockets: connections
  })
}

/** @type {http.RequestOptions} */
const httpKeepAliveOptions = {
  ...httpBaseOptions,
  agent: new http.Agent({
    keepAlive: true,
    maxSockets: connections
  })
}

const axiosAgent = new http.Agent({
  keepAlive: true,
  maxSockets: connections
})

const fetchAgent = new http.Agent({
  keepAlive: true,
  maxSockets: connections
})

const gotAgent = new http.Agent({
  keepAlive: true,
  maxSockets: connections
})

const requestAgent = new http.Agent({
  keepAlive: true,
  maxSockets: connections
})

const superagentAgent = new http.Agent({
  keepAlive: true,
  maxSockets: connections
})

// https://github.com/ladjs/superagent/issues/1540#issue-561464561
const superagent = _superagent.agent().use((req) => req.agent(superagentAgent))

const undiciOptions = {
  path: '/',
  method: 'GET',
  blocking: false,
  reset: false,
  headersTimeout,
  bodyTimeout
}

const Class = connections > 1 ? Pool : Client
const dispatcher = new Class(httpBaseOptions.url, {
  pipelining,
  connections,
  ...dest
})

const globalDispatcher = new Agent({
  pipelining,
  connections,
  connect: {
    rejectUnauthorized: false
  }
})

setGlobalDispatcher(globalDispatcher)

async function cleanup () {
  httpNoKeepAliveOptions.agent.destroy()
  httpKeepAliveOptions.agent.destroy()
  axiosAgent.destroy()
  fetchAgent.destroy()
  gotAgent.destroy()
  requestAgent.destroy()
  superagentAgent.destroy()

  await Promise.allSettled([dispatcher.destroy(), globalDispatcher.destroy()])
}

class SimpleRequest {
  constructor (resolve) {
    this.dst = new Writable({
      write (chunk, encoding, callback) {
        callback()
      }
    }).on('finish', resolve)
  }

  onRequestStart (controller) {
    this.controller = controller
  }

  onResponseStart (controller, statusCode, headers, statusText) {
    this.dst.on('drain', () => controller.resume())
  }

  onResponseData (controller, chunk) {
    if (this.dst.write(chunk) === false) {
      controller.pause()
    }
  }

  onResponseEnd (controller, trailers) {
    this.dst.end()
  }

  onResponseError (controller, err) {
    throw err
  }
}

const experiments = {
  'http - no keepalive' () {
    return makeParallelRequests(resolve => {
      http.get(httpNoKeepAliveOptions, res => {
        res
          .pipe(
            new Writable({
              write (chunk, encoding, callback) {
                callback()
              }
            })
          )
          .on('finish', resolve)
      })
    })
  },
  'http - keepalive' () {
    return makeParallelRequests(resolve => {
      http.get(httpKeepAliveOptions, res => {
        res
          .pipe(
            new Writable({
              write (chunk, encoding, callback) {
                callback()
              }
            })
          )
          .on('finish', resolve)
      })
    })
  },
  'undici - pipeline' () {
    return makeParallelRequests(resolve => {
      dispatcher
        .pipeline(undiciOptions, ({ body }) => {
          return body
        })
        .end()
        .pipe(
          new Writable({
            write (chunk, encoding, callback) {
              callback()
            }
          })
        )
        .on('finish', resolve)
    })
  },
  'undici - request' () {
    return makeParallelRequests(resolve => {
      dispatcher.request(undiciOptions).then(({ body }) => {
        body
          .pipe(
            new Writable({
              write (chunk, encoding, callback) {
                callback()
              }
            })
          )
          .on('finish', resolve)
      })
    })
  },
  'undici - stream' () {
    return makeParallelRequests(resolve => {
      return dispatcher
        .stream(undiciOptions, () => {
          return new Writable({
            write (chunk, encoding, callback) {
              callback()
            }
          })
        })
        .then(resolve)
    })
  },
  'undici - dispatch' () {
    return makeParallelRequests(resolve => {
      dispatcher.dispatch(undiciOptions, new SimpleRequest(resolve))
    })
  }
}

if (process.env.PORT) {
  // fetch does not support the socket
  experiments['undici - fetch'] = () => {
    return makeParallelRequests(resolve => {
      fetch(dest.url)
        .then(res => {
          res.body.pipeTo(
            new WritableStream({
              write () {},
              close () {
                resolve()
              }
            })
          )
        })
        .catch(console.log)
    })
  }

  experiments['node-fetch'] = () => {
    return makeParallelRequests(resolve => {
      nodeFetch(dest.url, { agent: fetchAgent })
        .then(res => {
          res.body
            .pipe(
              new Writable({
                write (chunk, encoding, callback) {
                  callback()
                }
              })
            )
            .on('finish', resolve)
        })
        .catch(console.log)
    })
  }

  const axiosOptions = {
    url: dest.url,
    method: 'GET',
    responseType: 'stream',
    httpAgent: axiosAgent
  }
  experiments.axios = () => {
    return makeParallelRequests(resolve => {
      axios
        .request(axiosOptions)
        .then(res => {
          res.data
            .pipe(
              new Writable({
                write (chunk, encoding, callback) {
                  callback()
                }
              })
            )
            .on('finish', resolve)
        })
        .catch(console.log)
    })
  }

  const gotOptions = {
    method: 'GET',
    agent: {
      http: gotAgent
    }
  }
  experiments.got = () => {
    return makeParallelRequests(resolve => {
      got.stream(dest.url, gotOptions).pipe(new Writable({
        write (chunk, encoding, callback) {
          callback()
        }
      })).on('finish', resolve)
    })
  }

  const requestOptions = {
    url: dest.url,
    method: 'GET',
    agent: requestAgent,
    // avoid body toString
    encoding: null
  }
  experiments.request = () => {
    return makeParallelRequests(resolve => {
      request(requestOptions)
        .then(() => {
          // already body consumed
          resolve()
        })
        .catch(console.log)
    })
  }

  experiments.superagent = () => {
    return makeParallelRequests(resolve => {
      superagent
        .get(dest.url)
        .pipe(
          new Writable({
            write (chunk, encoding, callback) {
              callback()
            }
          })
        )
        .on('finish', resolve)
    })
  }
}

function main () {
  // cronometro runs each benchmark in a worker, so attach cleanup to every
  // test before the worker exits.
  const tests = Object.fromEntries(
    Object.entries(experiments).map(([name, test]) => {
      return [name, { test, after: cleanup }]
    })
  )

  cronometro(
    tests,
    {
      iterations,
      errorThreshold,
      print: false
    },
    (err, results) => {
      if (err) {
        throw err
      }

      printResults(results)
      cleanup()
    }
  )
}

if (isMainThread) {
  main()
} else {
  module.exports = main
}