--[[----------------------------------------------------------------------------

WFRxHttpClientImp.lua

--------------------------------------------------------------------------------

ADOBE CONFIDENTIAL
------------------
Copyright 2012 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

--------------------------------------------------------------------------------

This is the thread that manages access to the native implementation for WFRxHttpClient.
It should be accessed only by the public API class WFRxHttpClient.

------------------------------------------------------------------------------]]

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.lua.threading
local WFRequestDispatcher = require 'WFRequestDispatcher'

	-- WF.core
local KSDate = require 'KSDate'
local WFConfiguration = require 'WFConfiguration'
local WFQueueing = require 'WFQueueing'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFCapabilities = require 'WFCapabilities'

	-- WF.web
local WFRxRateLimiter = require 'WFRxRateLimiter'

	-- WF.core.threading
local WFCurrentThread = require 'WFCurrentThread'

	-- 
local WFPlatformInfo = require 'WFPlatformInfo'

	-- Configurator
local ENABLE_USE_LIBCURL_HTTP_CLIENT = ( WFConfiguration.Shared.disableLibcurlHttpClient ~= true )
local ENABLE_HTTP_LOGGING = WFConfiguration.Shared.enableHttpLogging
if not WFPlatformInfo.isDesktop then
	ENABLE_HTTP_LOGGING = WFCapabilities.getCapability( "WFRxHttpClientImp.enableHttpResponseLog" )
end

local KSTableUtils

	-- WF.web.http_client
local useCurl = WFConfiguration.Http.useCurl or (WFPlatformInfo.isDesktop and ENABLE_USE_LIBCURL_HTTP_CLIENT)
local WFHttpClientImp_native
if useCurl then
	WFHttpClientImp_native = require 'WFHttpClientImp_curl'
else
	WFHttpClientImp_native = require 'WFHttpClientImp_native'
end

local WFHttpClientHeaderParser = require 'WFHttpClientHeaderParser'

--------------------------------------------------------------------------------

local httpConnectionStats = WFDiagnosticLog.makeRingBuffer {
	title = "Number of HTTP connections",
	accumulate = true,
	"GET", "PUT"
}

--------------------------------------------------------------------------------

local kMaxNumSimultaneousRequests = WFConfiguration.Nimbus.maxNumSimultaneousHttpRequests or 10
	-- The number of slots in the HTTP rate limiter to allow at any given time. Different types
	-- of requests take up varying numbers of slots in the rate limiter.

local kRequestPriorityHighest = 5

local now = KSDate.timeNowForAnimation

local kTimings = setmetatable( {}, { __mode = 'kv' } )

local httpTimingStats = WFDiagnosticLog.makeRingBuffer {
	title = "Native HTTP client timings",
	"GET", "PUT"
}

local rateLimiter = WFRxRateLimiter.makeRateLimiter(
		WFQueueing.makePriorityQueueNew,
		kMaxNumSimultaneousRequests,
		kRequestPriorityHighest -- allow a single request when using the highest priority
		)

--------------------------------------------------------------------------------

local _errorObserver

--------------------------------------------------------------------------------

local logger = WFLogger 'WFRxHttpClientImp'
--	:setLevel 'debug'

local kMessageType_Progress = 'progress'
local kMessageType_Result = 'result'
local kMessageType_RequestInProgress = 'requestInProgress'

local timeNow = assert( KSDate.timeNow, "KSDate.timeNow not found" )

--------------------------------------------------------------------------------

local sUserAgent

local function normalizeAppNumber( appNumber )

	-- Ensure the appNumber starts with two numbers, separated by a dot
	-- This is required for Oz to recognize us.

	if not string.find( appNumber, "^%d+%.%d+%." ) then
		appNumber = string.gsub( appNumber, "^%d+%.%d+", "%0.0" )
	end
	if not string.find( appNumber, "^%d+%.%d+%." ) then
		appNumber = string.gsub( appNumber, "^%d+", "%0.0.0" )
	end
	if not string.find( appNumber, "^%d+%.%d+%." ) then
		appNumber = "0.0." .. appNumber
	end

	return appNumber

end

--------------------------------------------------------------------------------

local function getUserAgent()

	if not sUserAgent then

		-- Configure default user agent.
		-- Clients other than Revel will need to add their appIdentifier
		-- to the case below if they want a custom user agent name.

		local function optFn( itemName, default )

			local appBuildNum = WFCapabilities.getCapability( 'WFHttpClient_' .. itemName )
			
			if not appBuildNum then

				local fn = WFPlatformInfo[ 'get' .. itemName ]

				appBuildNum = fn and fn()

				if not appBuildNum then
					appBuildNum = default
				end

			end

			return appBuildNum

		end
		
		local appNumber
		if WFPlatformInfo.isAndroid then
			appNumber = optFn( 'AppVersion' )
			if not appNumber then
				appNumber = optFn( 'AppVersion', "0.0" ) .. ".Dev"
			end
		else
		local version = optFn( 'AppVersion', '0.0' )
		
		local build = optFn( 'Build', 'Dev' )
	
			appNumber = string.format( "%s.%s", version, build )
		end
		
		appNumber = normalizeAppNumber( appNumber )

		local baseName = optFn( 'UserAgentAppName' )
		if not baseName then
			baseName = optFn( 'AppName', 'Wichita Foundation Client' )
		end

		local manufacturer = optFn( 'DeviceManufacturer', 'Unknown' )
		local model = optFn( 'DeviceModel', 'Unknown' )

		local mapWFPlatformOSToUserAgentOS = {
			MacOSX = "OSX",
			[ "Windows 7" ] = "Windows",
--			Android = "Android",
--			iOS = "iOS",
--			Windows = "Windows",
--			Unknown = "Unknown",
		}

		local os = WFPlatformInfo.getOS()
		os = mapWFPlatformOSToUserAgentOS[ os ] or os

		local osVer = optFn( 'OSVersion', "?.?.?" )

		sUserAgent = string.format(
			"%s(%s;%s;%s;%s;%s)",
			baseName,
			appNumber,
			manufacturer,
			model,
			os,
			osVer )
	end

	return sUserAgent

end

--============================================================================--

-- @mark httpRequestSherpa (internal to this file)

local sherpaForRequest = {}
	-- key (native object): the native HTTP request object (only responds to :close)
	-- value (httpRequestSherpa instance)

local httpRequestSherpaMethods = {

	_httpRequest = false,
		-- (native object) the native HTTP request object (only responds to :close)

	_url = false,
		-- (string) URL we're fetching or putting to

	_options = false,
		-- (table) options passed to client thread get or put function

	_optData = false,
		-- (KSData or number) data to pass with request or total size if multipart post

	_performanceMetrics = false,
		-- (table) a table of performance metrics of the request, to send via WFPerformanceReport

	_headers = false,
		-- (table) headers received (they come in before data )

	_observer = false,
		-- observer that subscribed to this.

	_observable = false,
		-- Observable< what, ... > an observable for sending info back to the client.
		-- If what is "progress", then the ... = nBytesSoFar, nBytesExpected
		-- If what is "result", then ... = data

}

--------------------------------------------------------------------------------

function httpRequestSherpaMethods:__init( url, options, optData, optPerformanceMetrics )

    -- Call getCapability here because on mobile platforms setCapability is not
	-- called prior to this file loading.
	ENABLE_HTTP_LOGGING = ENABLE_HTTP_LOGGING or WFCapabilities.getCapability( "WFRxHttpClientImp.enableHttpLogging" )

	if ENABLE_HTTP_LOGGING then
		logger:configure { logLevel = 'trace', action = 'logfile' }
	end

	self._url = url

	self._options = options
	if useCurl then
		self._options.loggerLevel = logger._priority
	else
		self._options.bandwidthRateLimit = nil
	end
	self._optData = optData or false
		-- If we don't put the "or false" clause in, then we error out with
		-- illegal read on this field later.

	httpConnectionStats:add( optData and 0 or 1, optData and 1 or 0 )
	kTimings[ self ] = now()

	self._observable = Observable.create( function( observer )

		assert( not self._observer, "Cannot subscribe multiple times" )
		
		self._observer = observer

		self._performanceMetrics = optPerformanceMetrics or {}
	
		self._performanceMetrics.startTime = now()

		local reqOrError, reason, msg = WFHttpClientImp_native.getOrPut( self._url,
															 self._options,
															 self._optData or nil )

		if reqOrError == 'error' then

			-- An error here means something was wrong with the parameters so log it.
			logger:errorf( "getOrPut returned error reason=%s msg=%s", reason, msg )

			self._observer:onError( reason, msg )

		else

			self._httpRequest = reqOrError

			sherpaForRequest[ self._httpRequest ] = self

		end

		if not ENABLE_HTTP_LOGGING then
			if self._optData then

				local optDataType = type( self._optData )

				if optDataType ~= 'string' and optDataType ~= 'number' then
					-- release memory used by data ASAP
					self._optData:close()
				end

				self._optData = '< closed >'

			end
		end

		return function()
			self:close()
		end

	end )

end

--------------------------------------------------------------------------------

function httpRequestSherpaMethods:close()

	-- Close off the HTTP request.

	httpConnectionStats:add( self._optData and 0 or -1, self._optData and -1 or 0 )
	local time = now() - kTimings[ self ]
	kTimings[ self ] = nil
	httpTimingStats:add( self._optData and 0 or time, self._optData and time or 0 )

	local request = self._httpRequest

	logger:debugf( "Completed in %f s ( url = %s, options = %s )", time, self._url, self._options )

	if request then

		self._httpRequest = false
		request:close()

		sherpaForRequest[ request ] = nil

	end

	self._observer = '< closed >'
	self._observable = '< closed >'

end

--------------------------------------------------------------------------------

local HttpRequestSherpa = Rx.Class 'HttpRequestSherpa'( httpRequestSherpaMethods )

--------------------------------------------------------------------------------

-- This function gets triggered from native code when it has data to report
-- for any of the outstanding requests.

 local function completionHandler( request, status, data, optData )

	local sherpa = sherpaForRequest[ request ]

	if not sherpa then
		local errorMsg = "Completion handler got called for a request that we don't recognize"
		logger:warnf( errorMsg )
		return errorMsg
	end

	if status == 'header' then

		WFHttpClientHeaderParser.standardizeHeaders( data )
		sherpa._headers = data
			-- Save headers off; we'll include them as part of the final response.

	elseif status == 'data' then

		if ENABLE_HTTP_LOGGING then

			KSTableUtils = KSTableUtils or require "KSTableUtils"

			local logEntryOptions = KSTableUtils.deepCopy( sherpa._options )
			for _, v in ipairs( logEntryOptions.requestHeaders or {} ) do
				if v.header == "Authorization" then
					v.value = "-"
				end
			end

			if string.find( sherpa._url, "photos%-stage%.adobe%.io" ) or string.find( sherpa._url, "photos%.adobe%.io" ) then
				if sherpa._optData then
					logger:tracef( "Http url = %s, options = %s sent = %s response = %s, response_headers = %s )", sherpa._url, logEntryOptions, tostring(sherpa._optData), tostring(data), sherpa._headers )
				else
					logger:tracef( "Http url = %s, options = %s response = %s, response_headers = %s )", sherpa._url, logEntryOptions, tostring(data), sherpa._headers )
				end
			elseif string.find( sherpa._url, "storage%.adbephotos%-stage%.com") or string.find( sherpa._url, "storage%.adbephotos%.com") then
				-- don't log out binary file response
				logger:tracef( "Http url = %s, options = %s response_headers = %s )", sherpa._url, logEntryOptions, sherpa._headers )
			end

			if sherpa._optData then

				local optDataType = type( sherpa._optData )

				if optDataType ~= 'string' and optDataType ~= 'number' then
					sherpa._optData:close();
				end

				sherpa._optData = '< closed >'

			end

		end

		local headers = sherpa._headers

		if not headers then
			logger:error( "Huh ... got data but not headers!" )
		elseif headers.status_code >= '400' then
			logger:infof( "HTTP error on %s: \n\n%s\n\n%s", sherpa._url, headers, data )
		end

		if headers then
			
			local timing = kTimings[ sherpa ]
			
			if timing then
				headers[ 'wf_time' ] = now() - timing
			end

		end
		
		if headers and headers.status_code < '400' then	
			sherpa._observer:onNext( 'result', 'data', data, headers )
		else
			sherpa._observer:onNext( 'result', 'error', data, headers )
			if _errorObserver then
				_errorObserver:onNext( data, headers )
			end
		end

		sherpa._observer:onCompleted()

		if data and type( data ) ~= 'string' then
		
			WFRequestDispatcher.deferredClose( data )
				-- Close the KSData object after processing the current request.
				-- If we call deferredClose before the :onNext statements above,
				-- we risk closing the object _before_ we fire those data events.
				-- This was seen in bug WF-1294 on the _errorObserver stream.
				-- Just like :onNext this also applies to :onCompleted()
		end

		-- Performance Reporting
		
		local performanceMetrics = sherpa._performanceMetrics or {}
		
		if performanceMetrics.startTime then
			performanceMetrics.requestDuration = now() - performanceMetrics.startTime
			WFPerformanceReport.recordSingleValue( 'WFRxHttpClientImp: getOrPut', performanceMetrics )
		end
		
	elseif status == 'progress' then

		sherpa._observer:onNext( 'progress', data, optData )	-- really nBytesSoFar, nBytesExpected

	elseif status == 'error' then

		if ENABLE_HTTP_LOGGING then
			logger:tracef( "Http url = %s, options = %s error = %s, %s )", sherpa._url, sherpa._options, tostring(data), tostring(optData) )
		end

		sherpa._observer:onNext( 'result', 'error', data, nil, optData )
		sherpa._observer:onCompleted()
		
		if _errorObserver then
			_errorObserver:onNext( data, nil, optData )
		end

	elseif status == 'log' then

		local output = "NativeImpl: " .. optData
		if data == 'warn' then
			logger:warnf( output )
		elseif data == 'trace' then
			logger:tracef( output )
		elseif data == 'debug' then
			logger:debugf( output )
		elseif data == 'info' then
			logger:infof( output )
		elseif data == 'error' then
			logger:errorf( output )
		elseif data == 'fatal' then
			logger:fatalf( output )
		else
			logger:infof( output )
		end

	else

		local errorMsg = string.format( "Bogus status from native code: %s", status )
		logger:errorf( errorMsg )
		return errorMsg

	end

	return true

end

WFHttpClientImp_native.registerCompletionHandler( completionHandler )

--------------------------------------------------------------------------------

-- Return true if the headers specified by these options include a User-Agent.
-- Also return the header array, creating one in options.requestHeaders if necessary.

local function hasUserAgentHeader( options )

	local headers = options.requestHeaders

	if not headers then
		headers = {}
		options.requestHeaders = headers
	end

	local hasUserAgent = false

	for _, h in ipairs( headers ) do
		if h.header == 'User-Agent' then
			hasUserAgent = true
		end
	end

	return hasUserAgent, headers

end

--------------------------------------------------------------------------------

local offsetError = "Error, offset greater than length"

local function downgradeDataForLegacyNativeImplementations( options, data )

	local KSData = require 'KSData'

	local usingFilePath = options.dataFormat == 'filePath'
	if usingFilePath then

		data = KSData.memoryMappedContentsOfFile( data )
			-- GRUNGE [ arahn 2012-08-22]: Using MMAP means we can't relocate this
			-- file on Windows so long as the KSData is alive.	We must remember
			-- to close this before trying to move the file!
	end

	if ( options.offset and options.offset > 1 ) or options.length then

		do
			local dataStr = tostring( data )
			local offset = options.offset or 1
			local length = options.length or ( #dataStr - offset + 1 )
            if length < 0 then
				if usingFilePath then
					data:close()
				end
			    error( offsetError, 2 )
			end

			local endpos = offset + length - 1
			dataStr = string.sub( dataStr, offset, endpos )
			data = KSData.make( dataStr )
		end

		-- Ensure that temp lua string is cleaned up
		collectgarbage "collect"
		collectgarbage "collect"
		collectgarbage "collect"

	end

	options.dataFormat = 'data'
	options.offset = nil
	options.length = nil

	return data

end

--------------------------------------------------------------------------------

local function elideFirstProgressEvent( messageType, extraProgressInfo, ... )

	-- The first progress event for an HTTP request has proved to be very
	-- innacurate, so we skip it. In the event that the HTTP request only
	-- had the one progress event, we replay it within gatherProgressData.

	if messageType == kMessageType_Progress then

		local _, totalBytes = ...

		extraProgressInfo.totalBytes = totalBytes

		extraProgressInfo.eventCount = extraProgressInfo.eventCount
				and extraProgressInfo.eventCount + 1 or 1

		if extraProgressInfo.eventCount == 1 then

			return false

		end

	end

	return true

end

--------------------------------------------------------------------------------

local function gatherProgressData( messageType, extraProgressInfo, ... )

	extraProgressInfo.timestamp = timeNow()

	if messageType == kMessageType_RequestInProgress then

		if not extraProgressInfo.startTime then
			extraProgressInfo.startTime = timeNow()
		end

	elseif messageType == kMessageType_Result then

		if not extraProgressInfo.endTime then
			extraProgressInfo.endTime = timeNow()
		end

		if extraProgressInfo.eventCount == 1 then

			-- We ignored the first progress event because the first event is always
			-- pretty innacurate, but there was only one event. In this case we still
			-- need to provide the bytes-per-second, which we do using the request start
			-- and end times.

			local reissuedProgressEvent  = {
				startTime = extraProgressInfo.startTime,
				timestamp = extraProgressInfo.timestamp
			}

			return Observable.just( kMessageType_Progress, reissuedProgressEvent, extraProgressInfo.totalBytes, extraProgressInfo.totalBytes )
				:concat( Observable.just( kMessageType_Result, extraProgressInfo, ... ) )

		end

	end

	return Observable.just( messageType, extraProgressInfo, ... )

end

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'getOrPut', function( url, options, optData )

	if not WFHttpClientImp_native.supportsFileAndOffset or options.forceLegacyFileAndOffset then

		logger:debug( "warning using downgradeDataForLegacyNativeImplementations" )

		local success
		success, optData = pcall(downgradeDataForLegacyNativeImplementations, options, optData )

		if not success then

			if optData == offsetError then
				return Observable.throw( offsetError )
            end

			return Observable.throw( 'fileio', optData )
		end

	end

	local hasUserAgent, headers = hasUserAgentHeader( options )
	if not hasUserAgent then
		headers[ #headers + 1 ] = { header = 'User-Agent', value = getUserAgent() }
	end

	logger:debugf( "getOrPut( url = %s, options = %s )", url, options )
	
	local requestPriority = options.requestPriority
	local requestCost = options.requestCost
	local bypassRateLimiter = options.bypassRateLimiter or false
	local requestDescription = options.requestDescription or 'Unknown'
	local notifyWhenHttpRequestIsInProgress = options.notifyWhenHttpRequestIsInProgress or false
	
	local maybeRateLimitedObservable
	
	local perfTable = {
		url = url,
		requestMethod = options.requestMethod,
		requestPriority = requestPriority,
		requestCost = requestCost,
	}

	if not bypassRateLimiter then
	
		-- Use a flatMap so we can re-issue a new request
		-- if the rate limited observable needs to be rescheduled.
		-- This is neccessary because the HTTP sherpa observables
		-- cannot be restarted by re-subscribing to them.
	
		local observableToLimit = Observable.just( true )
			:flatMap( function()
				local sherpa = HttpRequestSherpa( url, options or {}, optData, perfTable )
				return sherpa._observable
			end )
		
		if notifyWhenHttpRequestIsInProgress then
		
			observableToLimit = observableToLimit:startWith( kMessageType_RequestInProgress )

		end
			
		maybeRateLimitedObservable = rateLimiter:enqueueObservable( 
				observableToLimit, requestPriority, requestCost, requestDescription )
				
	else
	
		local sherpa = HttpRequestSherpa( url, options or {}, optData, perfTable )
		maybeRateLimitedObservable = sherpa._observable
	
		if notifyWhenHttpRequestIsInProgress then
		
			maybeRateLimitedObservable = maybeRateLimitedObservable:startWith( kMessageType_RequestInProgress )

		end
	
	end

	local extraProgressInfo = {}

	return maybeRateLimitedObservable

		:map( function( messageType, ... )
			return messageType, extraProgressInfo, ...
		end )

		:filter( elideFirstProgressEvent )

		:flatMap( function( ... )

			return options.sendProgress
					and gatherProgressData( ... )
					or Observable.just( ... )

		end )

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'setProxyCredentials', function( user, password )
		
	if WFHttpClientImp_native.setProxyCredentials then
		WFHttpClientImp_native.setProxyCredentials( user, password )
	end

	logger:debug( "setProxyCredentials( user = %s )", user )
	return true

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'watchHttpErrors', function()
		
	if not _errorObserver then
		_errorObserver = Rx.Subject()
	end

	return _errorObserver

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'setUserAgentAppName', function( appName )

	WFCapabilities.setCapability( "WFHttpClient_UserAgentAppName", appName );
	sUserAgent = nil

	return true

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'DEBUG_dumpCoverageP', function( path )

	require 'SetupCoverage'.dumpCoverage( path )
	return true

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'DEBUG_normalizeAppNumber', function( appNumber )

	return Observable.just( normalizeAppNumber( appNumber ) )

end )

--------------------------------------------------------------------------------

WFRequestDispatcher.onMessage( 'DEBUG_completionHandler', function( create, status, data, optData )

	local debugRequest = 'DebugSherpaRequest'

	if create then

		local mockDebugSherpa = {}
		for k, v in pairs(httpRequestSherpaMethods) do
			mockDebugSherpa[k] = v
		end
		mockDebugSherpa._url = "MockURL"
		mockDebugSherpa._options = {}
		
		mockDebugSherpa._observable = Observable.create( function( observer )
			mockDebugSherpa._observer = observer
			return function()
				sherpaForRequest[ debugRequest ] = nil	
			end
		end )

		sherpaForRequest[ debugRequest ] = mockDebugSherpa

		return mockDebugSherpa._observable

	else

		return completionHandler( debugRequest, status, data, optData )

	end

end )

--------------------------------------------------------------------------------

WFDiagnosticLog.addLogSectionFunc( function()

	local KSTableUtils = require "KSTableUtils"

	local threadName = WFCurrentThread.DEBUG_getThreadName() or '( unknown )'

	local activeRequests = {}
	for _, httpRequest in pairs( sherpaForRequest ) do

		local startTime = httpRequest and httpRequest._performanceMetrics and httpRequest._performanceMetrics.startTime

		local logEntryOptions = KSTableUtils.deepCopy( httpRequest._options )

		local logEntry = {
			url = httpRequest._url,
			options = logEntryOptions,
			currentDuration = startTime and now() - startTime,
			observerState = httpRequest._observer:isDisposed() and "disposed" or "active",
		}

		for _, v in ipairs( logEntryOptions.requestHeaders or {} ) do
			if v.header == "Authorization" then
				v.value = "-"
			end
		end

		table.insert( activeRequests, logEntry )

	end

	return {

		name = "WFRxHttpClientImp: " .. threadName,
		id = 'WFRxHttpClientImp',
		weight = 401,
		outputForAutomation = true,
		parentSection = WFDiagnosticLog.Sections.HTTP,
		
		body = {
			activeRequests = activeRequests,
			connections = { httpConnectionStats:visualize "boxAndWhisker" },
			timings = { httpTimingStats:visualize "boxAndWhisker" },
		}
	}

end )
