node-bhttp/lib/bhttp.coffee

# FIXME: Force-lowercase user-supplied headers before merging them into the request?
# FIXME: Deep-merge query-string arguments between URL and argument?
# FIXME: Named arrays for multipart/form-data?
# FIXME: Are arrays of streams in `data` correctly recognized as being streams?

# Core modules
urlUtil = require "url"
querystring = require "querystring"
stream = require "stream"
http = require "http"
https = require "https"
util = require "util"

# Utility modules
Promise = require "bluebird"
_ = require "lodash"
S = require "string"
formFixArray = require "form-fix-array"
errors = require "errors"
debug = require("debug")
debugRequest = debug("bhttp:request")
debugResponse = debug("bhttp:response")
extend = require "extend"

# Other third-party modules
formData = require "form-data2"
concatStream = require "concat-stream"
toughCookie = require "tough-cookie"
streamLength = require "stream-length"
sink = require "through2-sink"
spy = require "through2-spy"

# For the version in the user agent, etc.
packageConfig = require "../package.json"

bhttpErrors = {}

# Error types

errors.create
	name: "bhttpError"
	scope: bhttpErrors

errors.create
	name: "ConflictingOptionsError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

errors.create
	name: "UnsupportedProtocolError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

errors.create
	name: "RedirectError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

errors.create
	name: "MultipartError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

errors.create
	name: "ConnectionTimeoutError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

errors.create
	name: "ResponseTimeoutError"
	parents: bhttpErrors.bhttpError
	scope: bhttpErrors

# Utility functions

ofTypes = (obj, types) ->
	match = false
	for type in types
		match = match or obj instanceof type
	return match

addErrorData = (err, request, response, requestState) ->
	err.request = request
	err.response = response
	err.requestState = requestState
	return err

isStream = (obj) ->
	obj? and (ofTypes(obj, [stream.Readable, stream.Duplex, stream.Transform]) or obj.hasOwnProperty("_bhttpStreamWrapper"))

# Middleware
# NOTE: requestState is an object that signifies the current state of the overall request; eg. for a response involving one or more redirects, it will hold a 'redirect history'.
prepareSession = (request, response, requestState) ->
	debugRequest "preparing session"
	Promise.try ->
		if requestState.sessionOptions?
			# Request options take priority over session options
			request.options = _.merge _.clone(requestState.sessionOptions), request.options

		# Create a headers parameter if it doesn't exist yet - we'll need to add some stuff to this later on
		# FIXME: We may need to do a deep-clone of other mutable options later on as well; otherwise, when getting a redirect in a session with pre-defined options, the contents may not be correctly cleared after following the redirect.
		if request.options.headers?
			request.options.headers = _.clone(request.options.headers, true)
		else
			request.options.headers = {}

		# If we have a cookie jar, start out by setting the cookie string.
		if request.options.cookieJar?
			Promise.try ->
				# Move the cookieJar to the request object, the http/https module doesn't need it.
				request.cookieJar = request.options.cookieJar
				delete request.options.cookieJar

				# Get the current cookie string for the URL
				request.cookieJar.get request.url
			.then (cookieString) ->
				debugRequest "sending cookie string: %s", cookieString
				request.options.headers["cookie"] = cookieString
				Promise.resolve [request, response, requestState]
		else
			Promise.resolve [request, response, requestState]

prepareDefaults = (request, response, requestState) ->
	debugRequest "preparing defaults"
	Promise.try ->
		# These are the options that we need for response processing, but don't need to be passed on to the http/https module.
		request.responseOptions =
			discardResponse: request.options.discardResponse ? false
			keepRedirectResponses: request.options.keepRedirectResponses ? false
			followRedirects: request.options.followRedirects ? true
			noDecode: request.options.noDecode ? false
			decodeJSON: request.options.decodeJSON ? false
			stream: request.options.stream ? false
			justPrepare: request.options.justPrepare ? false
			redirectLimit: request.options.redirectLimit ? 10
			onDownloadProgress: request.options.onDownloadProgress
			responseTimeout: request.options.responseTimeout

		# Whether chunked transfer encoding for multipart/form-data payloads is acceptable. This is likely to break quietly on a lot of servers.
		request.options.allowChunkedMultipart ?= false

		# Whether we should always use multipart/form-data for payloads, even if querystring-encoding would be a possibility.
		request.options.forceMultipart ?= false

		# If no custom user-agent is defined, set our own
		request.options.headers["user-agent"] ?= "bhttp/#{packageConfig.version}"

		# Normalize the request method to lowercase.
		request.options.method = request.options.method.toLowerCase()

		Promise.resolve [request, response, requestState]

prepareUrl = (request, response, requestState) ->
	debugRequest "preparing URL"
	Promise.try ->
		# Parse the specified URL, and use the resulting information to build a complete `options` object
		urlOptions = urlUtil.parse request.url, true

		_.extend request.options, {hostname: urlOptions.hostname, port: urlOptions.port}
		request.options.path = urlUtil.format {pathname: urlOptions.pathname, query: request.options.query ? urlOptions.query}
		request.protocol = S(urlOptions.protocol).chompRight(":").toString()

		Promise.resolve [request, response, requestState]

prepareProtocol = (request, response, requestState) ->
	debugRequest "preparing protocol"
	Promise.try ->
		request.protocolModule = switch request.protocol
			when "http" then http
			when "https" then https # CAUTION / FIXME: Node will silently ignore SSL settings without a custom agent!
			else null

		if not request.protocolModule?
			return Promise.reject() new bhttpErrors.UnsupportedProtocolError "The protocol specified (#{protocol}) is not currently supported by this module."

		request.options.port ?= switch request.protocol
			when "http" then 80
			when "https" then 443

		Promise.resolve [request, response, requestState]

prepareOptions = (request, response, requestState) ->
	debugRequest "preparing options"
	Promise.try ->
		# Do some sanity checks - there are a number of options that cannot be used together
		if (request.options.formFields? or request.options.files?) and (request.options.inputStream? or request.options.inputBuffer?)
			return Promise.reject addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot define both formFields/files and a raw inputStream or inputBuffer."), request, response, requestState)

		if request.options.encodeJSON and (request.options.inputStream? or request.options.inputBuffer?)
			return Promise.reject addErrorData(new bhttpErrors.ConflictingOptionsError("You cannot use both encodeJSON and a raw inputStream or inputBuffer.", undefined, "If you meant to JSON-encode the stream, you will currently have to do so manually."), request, response, requestState)

		# If the user plans on streaming the response, we need to disable the agent entirely - otherwise the streams will block the pool.
		if request.responseOptions.stream
			request.options.agent ?= false

		Promise.resolve [request, response, requestState]

preparePayload = (request, response, requestState) ->
	debugRequest "preparing payload"
	Promise.try ->
		# Persist the download progress event handler on the request object, if there is one.
		request.onUploadProgress = request.options.onUploadProgress

		# If a 'files' parameter is present, then we will send the form data as multipart data - it's most likely binary data.
		multipart = request.options.forceMultipart or request.options.files?

		# Similarly, if any of the formFields values are either a Stream or a Buffer, we will assume that the form should be sent as multipart.
		multipart = multipart or _.any request.options.formFields, (item) ->
			item instanceof Buffer or isStream(item)

		# Really, 'files' and 'formFields' are the same thing - they mostly have different names for 1) clarity and 2) multipart detection. We combine them here.
		_.extend request.options.formFields, request.options.files

		# For a last sanity check, we want to know whether there are any Stream objects in our form data *at all* - these can't be used when encodeJSON is enabled.
		containsStreams = _.any request.options.formFields, (item) -> isStream(item)

		if request.options.encodeJSON and containsStreams
			return Promise.reject() new bhttpErrors.ConflictingOptionsError "Sending a JSON-encoded payload containing data from a stream is not currently supported.", undefined, "Either don't use encodeJSON, or read your stream into a string or Buffer."

		if request.options.method in ["post", "put", "patch"]
			# Prepare the payload, and set the appropriate headers.
			if (request.options.encodeJSON or request.options.formFields?) and not multipart
				# We know the payload and its size in advance.
				debugRequest "got url-encodable form-data"
				request.options.headers["content-type"] = "application/x-www-form-urlencoded"

				if request.options.encodeJSON
					request.payload = JSON.stringify request.options.formFields ? null
				else if not _.isEmpty request.options.formFields
					# The `querystring` module copies the key name verbatim, even if the value is actually an array. Things like PHP don't understand this, and expect every array-containing key to be suffixed with []. We'll just append that ourselves, then.
					request.payload = querystring.stringify formFixArray(request.options.formFields)
				else
					request.payload = ""

				request.options.headers["content-length"] = request.payload.length

				return Promise.resolve()
			else if request.options.formFields? and multipart
				# This is going to be multipart data, and we'll let `form-data` set the headers for us.
				debugRequest "got multipart form-data"
				formDataObject = new formData()

				for fieldName, fieldValue of formFixArray(request.options.formFields)
					if not _.isArray fieldValue
						fieldValue = [fieldValue]

					for valueElement in fieldValue
						if valueElement._bhttpStreamWrapper?
							streamOptions = valueElement.options
							valueElement = valueElement.stream
						else
							streamOptions = {}

						formDataObject.append fieldName, valueElement, streamOptions

				request.payloadStream = formDataObject

				Promise.try ->
					formDataObject.getHeaders()
				.then (headers) ->
					if headers["content-transfer-encoding"] == "chunked" and not request.options.allowChunkedMultipart
						Promise.reject addErrorData(new MultipartError("Most servers do not support chunked transfer encoding for multipart/form-data payloads, and we could not determine the length of all the input streams. See the documentation for more information."), request, response, requestState)
					else
						_.extend request.options.headers, headers
						Promise.resolve()
			else if request.options.inputStream?
				# A raw inputStream was provided, just leave it be.
				debugRequest "got inputStream"
				Promise.try ->
					request.payloadStream = request.options.inputStream

					if request.payloadStream._bhttpStreamWrapper? and (request.payloadStream.options.contentLength? or request.payloadStream.options.knownLength?)
						Promise.resolve(request.payloadStream.options.contentLength ? request.payloadStream.options.knownLength)
					else
						streamLength request.options.inputStream
				.then (length) ->
					debugRequest "length for inputStream is %s", length
					request.options.headers["content-length"] = length
				.catch (err) ->
					debugRequest "unable to determine inputStream length, switching to chunked transfer encoding"
					request.options.headers["content-transfer-encoding"] = "chunked"
			else if request.options.inputBuffer?
				# A raw inputBuffer was provided, just leave it be (but make sure it's an actual Buffer).
				debugRequest "got inputBuffer"
				if typeof request.options.inputBuffer == "string"
					request.payload = new Buffer(request.options.inputBuffer) # Input string should be utf-8!
				else
					request.payload = request.options.inputBuffer

				debugRequest "length for inputBuffer is %s", request.payload.length
				request.options.headers["content-length"] = request.payload.length

				return Promise.resolve()
		else
			# GET, HEAD and DELETE should not have a payload. While technically not prohibited by the spec, it's also not specified, and we'd rather not upset poorly-compliant webservers.
			# FIXME: Should this throw an Error?
			return Promise.resolve()
	.then ->
		Promise.resolve [request, response, requestState]

prepareCleanup = (request, response, requestState) ->
	debugRequest "preparing cleanup"
	Promise.try ->
		# Remove the options that we're not going to pass on to the actual http/https library.
		delete request.options[key] for key in ["query", "formFields", "files", "encodeJSON", "inputStream", "inputBuffer", "discardResponse", "keepRedirectResponses", "followRedirects", "noDecode", "decodeJSON", "allowChunkedMultipart", "forceMultipart", "onUploadProgress", "onDownloadProgress"]

		# Lo-Dash apparently has no `map` equivalent for object keys...?
		fixedHeaders = {}
		for key, value of request.options.headers
			fixedHeaders[key.toLowerCase()] = value
		request.options.headers = fixedHeaders

		Promise.resolve [request, response, requestState]

# The guts of the module

prepareRequest = (request, response, requestState) ->
	debugRequest "preparing request"
	# FIXME: Mock httpd for testing functionality.
	Promise.try ->
		middlewareFunctions = [
			prepareSession
			prepareDefaults
			prepareUrl
			prepareProtocol
			prepareOptions
			preparePayload
			prepareCleanup
		]

		promiseChain = Promise.resolve [request, response, requestState]

		middlewareFunctions.forEach (middleware) -> # We must use the functional construct here, to avoid losing references
			promiseChain = promiseChain.spread (_request, _response, _requestState) ->
				middleware(_request, _response, _requestState)

		return promiseChain

makeRequest = (request, response, requestState) ->
	debugRequest "making %s request to %s", request.options.method.toUpperCase(), request.url
	Promise.try ->
		# Instantiate a regular HTTP/HTTPS request
		req = request.protocolModule.request request.options

		timeoutTimer = null

		new Promise (resolve, reject) ->
			# Connection timeout handling, if one is set.
			if request.responseOptions.responseTimeout?
				debugRequest "setting response timeout timer to #{request.responseOptions.responseTimeout}ms..."
				req.on "socket", (socket) ->
					timeoutHandler = ->
						debugRequest "a response timeout occurred!"
						req.abort()
						reject addErrorData(new bhttpErrors.ResponseTimeoutError("The response timed out."))

					timeoutTimer = setTimeout(timeoutHandler, request.responseOptions.responseTimeout)

			# Set up the upload progress monitoring.
			totalBytes = request.options.headers["content-length"]
			completedBytes = 0

			progressStream = spy (chunk) ->
				completedBytes += chunk.length
				req.emit "progress", completedBytes, totalBytes

			if request.onUploadProgress?
				req.on "progress", (completedBytes, totalBytes) ->
					request.onUploadProgress(completedBytes, totalBytes, req)

			# This is where we write our payload or stream to the request, and the actual request is made.
			if request.payload?
				# The entire payload is a single Buffer. We'll still pretend that it's a stream for our progress events, though, to provide a consistent API.
				debugRequest "sending payload"
				req.emit "progress", request.payload.length, request.payload.length
				req.write request.payload
				req.end()
			else if request.payloadStream?
				# The payload is a stream.
				debugRequest "piping payloadStream"
				if request.payloadStream._bhttpStreamWrapper?
					request.payloadStream.stream
						.pipe progressStream
						.pipe req
				else
					request.payloadStream
						.pipe progressStream
						.pipe req
			else
				# For GET, HEAD, DELETE, etc. there is no payload, but we still need to call end() to complete the request.
				debugRequest "closing request without payload"
				req.end()

			# In case something goes wrong during this process, somehow...
			req.on "error", (err) ->
				if err.code == "ETIMEDOUT"
					debugRequest "a connection timeout occurred!"
					reject addErrorData(new bhttpErrors.ConnectionTimeoutError("The connection timed out."))
				else
					reject err

			req.on "response", (res) ->
				if timeoutTimer?
					debugResponse "got response in time, clearing response timeout timer"
					clearTimeout(timeoutTimer)
				resolve res
	.then (response) ->
		Promise.resolve [request, response, requestState]

processResponse = (request, response, requestState) ->
	debugResponse "processing response, got status code %s", response.statusCode

	# When we receive the response, we'll buffer it up and/or decode it, depending on what the user specified, and resolve the returned Promise. If the user just wants the raw stream, we resolve immediately after receiving a response.

	Promise.try ->
		# First, if a cookie jar is set and we received one or more cookies from the server, we should store them in our cookieJar.
		if request.cookieJar? and response.headers["set-cookie"]?
			promises = for cookieHeader in response.headers["set-cookie"]
				debugResponse "storing cookie: %s", cookieHeader
				request.cookieJar.set cookieHeader, request.url
			Promise.all promises
		else
			Promise.resolve()
	.then ->
		# Now the actual response processing.
		response.request = request
		response.requestState = requestState
		response.redirectHistory = requestState.redirectHistory

		if response.statusCode in [301, 302, 303, 307] and request.responseOptions.followRedirects
			if requestState.redirectHistory.length >= (request.responseOptions.redirectLimit - 1)
				return Promise.reject addErrorData(new bhttpErrors.RedirectError("The maximum amount of redirects ({request.responseOptions.redirectLimit}) was reached."))

			# 301: For GET and HEAD, redirect unchanged. For POST, PUT, PATCH, DELETE, "ask user" (in our case: throw an error.)
			# 302: Redirect, change method to GET.
			# 303: Redirect, change method to GET.
			# 307: Redirect, retain method. Make same request again.
			switch response.statusCode
				when 301
					switch request.options.method
						when "get", "head"
							return redirectUnchanged request, response, requestState
						when "post", "put", "patch", "delete"
							return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 301 redirect for POST, PUT, PATCH or DELETE. RFC says we can't automatically continue."), request, response, requestState)
						else
							return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 301 redirect, but not sure how to proceed for the #{request.options.method.toUpperCase()} method."))
				when 302, 303
					return redirectGet request, response, requestState
				when 307
					if request.containsStreams and request.options.method not in ["get", "head"]
						return Promise.reject addErrorData(new bhttpErrors.RedirectError("Encountered a 307 redirect for POST, PUT or DELETE, but your payload contained (single-use) streams. We therefore can't automatically follow the redirect."), request, response, requestState)
					else
						return redirectUnchanged request, response, requestState
		else if request.responseOptions.discardResponse
			response.resume() # Drain the response stream
			Promise.resolve response
		else
			totalBytes = response.headers["content-length"]
			if totalBytes? # Otherwise `undefined` will turn into `NaN`, and we don't want that.
				totalBytes = parseInt(totalBytes)
			completedBytes = 0

			progressStream = sink (chunk) ->
				completedBytes += chunk.length
				response.emit "progress", completedBytes, totalBytes

			if request.responseOptions.onDownloadProgress?
				response.on "progress", (completedBytes, totalBytes) ->
					request.responseOptions.onDownloadProgress(completedBytes, totalBytes, response)

			new Promise (resolve, reject) ->
				# This is slightly hacky, but returning a .pipe'd stream would make all the response object attributes unavailable to the end user. Therefore, 'branching' the stream into our progress monitoring stream is a much better option. We use .pause() to ensure the stream doesn't start flowing until the end user (or our library) has actually piped it into something.
				response.pipe(progressStream)
				response.pause()

				if request.responseOptions.stream
					resolve response
				else
					response.on "error", (err) ->
						reject err

					response.pipe concatStream (body) ->
						# FIXME: Separate module for header parsing?
						if request.responseOptions.decodeJSON or ((response.headers["content-type"] ? "").split(";")[0] == "application/json" and not request.responseOptions.noDecode)
							try
								response.body = JSON.parse body
							catch err
								reject err
						else
							response.body = body

						resolve response

	.then (response) ->
		Promise.resolve [request, response, requestState]

# Some wrappers

doPayloadRequest = (url, data, options, callback) ->
	# A wrapper that processes the second argument to .post, .put, .patch shorthand API methods.
	# FIXME: Treat a {} for data as a null? Otherwise {} combined with inputBuffer/inputStream will error.
	if isStream(data)
		options.inputStream = data
	else if ofTypes(data, [Buffer]) or typeof data == "string"
		options.inputBuffer = data
	else
		options.formFields = data

	@request url, options, callback

redirectGet = (request, response, requestState) ->
	debugResponse "following forced-GET redirect to %s", response.headers["location"]
	Promise.try ->
		options = _.clone(requestState.originalOptions)
		options.method = "get"

		delete options[key] for key in ["inputBuffer", "inputStream", "files", "formFields"]

		doRedirect request, response, requestState, options

redirectUnchanged = (request, response, requestState) ->
	debugResponse "following same-method redirect to %s", response.headers["location"]
	Promise.try ->
		options = _.clone(requestState.originalOptions)
		doRedirect request, response, requestState, options

doRedirect = (request, response, requestState, newOptions) ->
	Promise.try ->
		if not request.responseOptions.keepRedirectResponses
			response.resume() # Let the response stream drain out...

		requestState.redirectHistory.push response
		bhttpAPI._doRequest response.headers["location"], newOptions, requestState

createCookieJar = (jar) ->
	# Creates a cookie jar wrapper with a simplified API.
	return {
		set: (cookie, url) ->
			new Promise (resolve, reject) =>
				@jar.setCookie cookie, url, (err, cookie) ->
					if err then reject(err) else resolve(cookie)
		get: (url) ->
			new Promise (resolve, reject) =>
				@jar.getCookieString url, (err, cookies) ->
					if err then reject(err) else resolve(cookies)
		jar: jar
	}

# The exposed API

bhttpAPI =
	head: (url, options = {}, callback) ->
		options.method = "head"
		@request url, options, callback
	get: (url, options = {}, callback) ->
		options.method = "get"
		@request url, options, callback
	post: (url, data, options = {}, callback) ->
		options.method = "post"
		doPayloadRequest.bind(this) url, data, options, callback
	put: (url, data, options = {}, callback) ->
		options.method = "put"
		doPayloadRequest.bind(this) url, data, options, callback
	patch: (url, data, options = {}, callback) ->
		options.method = "patch"
		doPayloadRequest.bind(this) url, data, options, callback
	delete: (url, data, options = {}, callback) ->
		options.method = "delete"
		@request url, options, callback
	request: (url, options = {}, callback) ->
		@_doRequest(url, options).nodeify(callback)
	_doRequest: (url, options, requestState) ->
		# This is split from the `request` method, so that the user doesn't have to pass in `undefined` for the `requestState` when they want to specify a `callback`.
		Promise.try =>
			request = {url: url, options: _.clone(options)}
			response = null
			requestState ?= {originalOptions: _.clone(options), redirectHistory: []}
			requestState.sessionOptions ?= @_sessionOptions ? {}

			prepareRequest request, response, requestState
		.spread (request, response, requestState) =>
			if request.responseOptions.justPrepare
				Promise.resolve [request, response, requestState]
			else
				Promise.try ->
					bhttpAPI.executeRequest request, response, requestState
				.spread (request, response, requestState) ->
					# The user likely only wants the response.
					Promise.resolve response
	executeRequest: (request, response, requestState) ->
		# Executes a pre-configured request.
		Promise.try ->
			makeRequest request, response, requestState
		.spread (request, response, requestState) ->
			processResponse request, response, requestState
	session: (options) ->
		options ?= {}
		options = _.clone options
		session = {}

		for key, value of this
			if value instanceof Function
				value = value.bind(session)
			session[key] = value

		if not options.cookieJar?
			options.cookieJar = createCookieJar(new toughCookie.CookieJar())
		else if options.cookieJar == false
			delete options.cookieJar
		else
			# Assume we've gotten a cookie jar.
			options.cookieJar = createCookieJar(options.cookieJar)

		session._sessionOptions = options

		return session
	wrapStream: (stream, options) ->
		# This is a method for wrapping a stream in an object that also contains metadata.
		return {
			_bhttpStreamWrapper: true
			stream: stream
			options: options
		}

extend(bhttpAPI, bhttpErrors)

module.exports = bhttpAPI

# That's all, folks!