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

WFCatDirectedChunkedUploader.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2021 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.

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

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

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFProgressReporter = require 'WFProgressReporter'

	-- WF.core.os
local WFDeviceInfo = require 'WFDeviceInfo'
local WFFileUtils = require 'WFFileUtils'

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'

	-- WF.web.utils
local WFURI = require 'WFURI'

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

local KSTableUtils_deepCopy = assert( KSTableUtils.deepCopy )
local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

local check = assert( WFArguments.check )

local logger = require 'WFLogger'( 'WFCatDirectChunkedUploader' )

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

local OZ_CHUNK_FILE_SIZE_RECOMMENDED = ( 5 * 1024 * 1024 )

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

local function maybeContentType( headers )

	local content_type
	for index, header in ipairs( headers or {} ) do
		if header and header.header == 'Content-Type' then
			content_type = header.value
			return content_type
		end
	end

	return content_type

end

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

local function calculateBytesNeedUpload( pendingParts )

	local bytesNeedUpload = 0

	for _, part in ipairs( pendingParts ) do

		bytesNeedUpload = bytesNeedUpload + ( part.byte_end - part.byte_start + 1 )

	end

	return bytesNeedUpload

end

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

local function getMainParts( self, mainPartsRequestUrl, fileSize, options, onHttpRequestInProgressCallbackFn, progressReporter )

	check( 'WFCatDirectChunkedUploader', 'string', 'number', 'table', '?function', '?WFProgressReporter' )

	local optionsCopy = KSTableUtils_deepCopy( options )

	optionsCopy.dataFormat = nil
	optionsCopy.includeProgress = nil
	optionsCopy.requestCost = 1

	optionsCopy.requestHeaders = optionsCopy.requestHeaders or {}
	optionsCopy.ignoreMockResponse = WFPlatformInfo.isUnitTest()
		-- This is necessary for certain CT-nightly tests.

	local contentType = maybeContentType( optionsCopy.requestHeaders )

	local _, mainPartsUrl = WFURI.replaceFromOzTemplate( mainPartsRequestUrl, {
		upload_status = 'pending',
		file_size = tostring( fileSize ),
		part_size = tostring( OZ_CHUNK_FILE_SIZE_RECOMMENDED ),
		content_type = WFURI.escape( contentType, 'http_query_key' ) 
	} )

	local requestInProgress = false

	local result = self._catalogSpec:get( mainPartsUrl, optionsCopy )

	if options.notifyWhenHttpRequestIsInProgress then

		result = result

			:tap( function( _, _, progress )

				if not requestInProgress and onHttpRequestInProgressCallbackFn then

					if progressReporter then
						progressReporter:setTotal( fileSize, true )
					end

					onHttpRequestInProgressCallbackFn()
					requestInProgress = true

				end

			end )

	end

	return result

		:takeLast( 1 )

		:flatMap( function( mainPartsResponseJson, headers )

			if type( headers ) ~= 'table' then

				return Observable.throw( "Fail to GET main/parts with unknown error (no headers)" )

			end

			if headers.status_code ~= '200' then

				local failureDescription = string.format( "Fail to GET main/parts with HTTP status code %s", headers.status_code or 'unknown' )

				return Observable.throw( failureDescription, headers )

			end

			-- decode urlDataJson into lua tables
			local ok, mainPartsOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, mainPartsResponseJson )

			if not ok then

				local failureDescription = string.format( "Fail to decode mainParts response error: %s", 
															mainPartsOrError and tostring( mainPartsOrError ) or "unknown" )

				return Observable.throw( failureDescription, headers )

			end

			return Observable.just( mainPartsOrError )

		end )

		:catch( function( responseData, headers )
			return Observable.throw( responseData, headers, mainPartsUrl )
		end )

end

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

local function putDirectChunk( self, url, filePath, options, startOffset, endOffset, fileSize, reportChunkProgress, onHttpRequestSuccessCallBackFn )

	local optionsCopy = KSTableUtils_deepCopy( options )

	-- Windows uses a long in string.format's implementation (which is 32bits on Windows)
	-- so use %s instead of %d to work around it.
	local directChunkedUploadStr = string.format( "bytes %s-%s/%s", startOffset, endOffset, fileSize )

	-- Offsets within this file are 0 based to make it easier to use the parts response.
	-- Add 1 as http's file offset is 1 based.
	optionsCopy.dataFormat = 'filePath'
	optionsCopy.offset = startOffset + 1
	optionsCopy.length = endOffset - startOffset + 1

	local description = startOffset == 0 and 'Initiate Direct Chunked Upload' or 'Uploading Direct Chunk'
	optionsCopy.requestDescription = string.format( '%s : %s %s', filePath, description, directChunkedUploadStr )

	optionsCopy.withoutAdobeAuthorization = true

	return self._catalogSpec:putBinary( url, filePath, optionsCopy )

		:tap( function( _, _, progress )

			if reportChunkProgress then

				reportChunkProgress( startOffset, progress )

			end

		end )

		:takeLast( 1 )

		:tap( onHttpRequestSuccessCallBackFn )

		:catch( function( responseData, headers )
			return Observable.throw( responseData, headers, url )
		end )

end

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

local function postCompleteUrl( self, completeUrl, options )

	local optionsCopy = KSTableUtils_deepCopy( options )

	optionsCopy.requestHeaders = optionsCopy.requestHeaders or {}

	optionsCopy.dataFormat = nil
	optionsCopy.notifyWhenHttpRequestIsInProgress = nil
	optionsCopy.requestCost = 1
	optionsCopy.bypassRateLimiter = true

	optionsCopy.requestMethod = 'POST'

	return self._catalogSpec:put( completeUrl, "", optionsCopy )

			:takeLast( 1 )

			:flatMap( function( data, headers )

				if type( headers ) ~= 'table' then

					return Observable.throw( "Fail to POST complete_upload with unknown error (no headers)" )

				end

				if headers.status_code ~= '202' then

					local errorMsg = string.format( "Fail to POST complete_upload request with HTTP status code %s",
									headers.status_code or 'unknown')

					logger:warnf( errorMsg )

					return Observable.throw( errorMsg, headers )

				end

				return Observable.just( true )

			end )

			:catch( function( responseData, headers )
				return Observable.throw( responseData, headers, completeUrl )
			end )

end

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

--[[!
	\class WFCatDirectChunkedUploader
]]--

local WFCatDirectChunkedUploader = {

	_catalogSpec = false,
		-- (WFCatSpec)

	_maxSimultaneous = false,
		-- (number) Number of simultaneous chunks to issue. The number relatively set to high for IOS as it uses the background process to upload

	WFCloseSupport.addCloseSupportMethods,

	private = {

		-- The following methods are internal, and are only exposed for testing.

		_getMainParts = getMainParts,

		_putDirectChunk = putDirectChunk,

		_postCompleteUrl = postCompleteUrl,
	}
}

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

function WFCatDirectChunkedUploader:__init( catalogSpec )

	check( 'WFCatDirectChunkedUploader', 'WFCatSpec' )

	self._catalogSpec = catalogSpec

	self._maxSimultaneous = WFPlatformInfo.isiOS and 100 or 5

end

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

function WFCatDirectChunkedUploader:doDirectChunkedUpload( mainPartsRequestUrl, filePath, developUserUpdated, options,
		onHttpRequestInProgressCallbackFn, onHttpRequestSuccessCallBackFn, ozExistenceCheckFn )

	check( 'WFCatDirectChunkedUploader', 'string', 'string', '?string', 'table', '?function', '?function', '?function' )
	
	local fileSize = WFFileUtils.fileAttributes( filePath ).fileSize or 0

	if fileSize == 0 then
		return Observable.throw( 'payload must exist', {} )
	end

	local progressReporter, progressStatusStream

	if options.includeProgress then
		progressReporter = WFProgressReporter( string.format( "doDirectChunkedUpload::%s", mainPartsRequestUrl ) )
		progressStatusStream = progressReporter:status()
	end

	local directChunkedUploadObservable = getMainParts( self, mainPartsRequestUrl, fileSize, options,
			onHttpRequestInProgressCallbackFn, progressReporter )

		:flatMap( function ( mainPartsResponse )

			local allParts = KSTableUtils_derefOpt( mainPartsResponse, 'parts' ) or {}
			local completeUrl = KSTableUtils_derefOpt( mainPartsResponse, 'links', 'rels/complete_upload', 'href' )

			-- some parts may have status uploaded, so they are not pending
			local pendingParts = {}
			for _, part in ipairs( allParts ) do
				if part.upload_status ~= 'uploaded' then
					pendingParts[ #pendingParts + 1 ] = part
				end
			end

			if completeUrl and developUserUpdated then

				local _, fromTemplate = WFURI.replaceFromOzTemplate( completeUrl, {
					device = WFURI.escape( WFDeviceInfo.deviceDescription(), 'http_query_key' ),
					userUpdated = WFURI.escape( developUserUpdated, 'http_query_key' )
				} )
				completeUrl = fromTemplate

			end

			if type(ozExistenceCheckFn) == 'function' and ozExistenceCheckFn( mainPartsResponse ) then

				return Observable.just( true )

			end

			if not completeUrl then

				return Observable.throw( "Missing complete upload URL" )

			end

			local bytesPreviouslyUploaded = 0

			if progressReporter and pendingParts then

				-- For resumed uploads, we don't want to include the number of bytes
				-- we previously uploaded in our bandwidth calculations. We keep track of
				-- the bytes previously uploaded.

				bytesPreviouslyUploaded = fileSize - calculateBytesNeedUpload( pendingParts )

				if bytesPreviouslyUploaded > 0 then

					progressReporter:setProcessedCount( bytesPreviouslyUploaded, true )

				end

			end

			local chunkedProgressInfo = {
				progressOfChunks = {}
			}

			local reportChunkProgress = function( startOffset, progress )

				if progressReporter and progress then

					chunkedProgressInfo.progressOfChunks[ startOffset ] = progress

					local bytesUploadedForRestOfChunks = 0

					for _, chunkProgress in pairs( chunkedProgressInfo.progressOfChunks ) do

						local bytesUploadedForChunk = chunkProgress.processedCount or 0
						bytesUploadedForRestOfChunks = bytesUploadedForRestOfChunks + bytesUploadedForChunk

					end

					local bytesUploaded = bytesPreviouslyUploaded + bytesUploadedForRestOfChunks
					progressReporter:setProcessedCount( bytesUploaded, nil, chunkedProgressInfo )

				end

			end

			local putChunksObservable = Observable.just( true )

			if next( pendingParts ) then

				putChunksObservable = Observable.from( pendingParts )

					:map( function( part )

						local putChunkUrl = KSTableUtils_derefOpt( part, 'href' )
						local startOffset = KSTableUtils_derefOpt( part, 'byte_start' )
						local endOffset = KSTableUtils_derefOpt( part, 'byte_end' )

						return putDirectChunk( self, putChunkUrl, filePath, options, startOffset, endOffset, fileSize, reportChunkProgress, onHttpRequestSuccessCallBackFn )

					end )

					:flatMerge( self._maxSimultaneous )

					:takeLast( 1 )

			end

			return putChunksObservable

				:flatMap( function()

					local baseUrl = KSTableUtils_derefOpt( mainPartsResponse, 'base' )
					local completeUploadUrl = WFURI.resolve( baseUrl, completeUrl )

					return postCompleteUrl( self, completeUploadUrl, options )

				end )

		end )

	local result = directChunkedUploadObservable

	if progressStatusStream then

		local packedChunkedUploadObservableStream = directChunkedUploadObservable

			:map( function( responseData, headers )
				return { responseData, headers }
			end )

			:finally( function()
				progressReporter:done()
			end )

			:startWith( {} )

		result = Observable.combineLatest( packedChunkedUploadObservableStream, progressStatusStream )

			:map( function( responseDataAndHeaders, ... )

				local responseData, headers = unpack( responseDataAndHeaders )

				return responseData, headers, ...

			end )

	end

	return result

end

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

return Rx.Class 'WFCatDirectChunkedUploader'( WFCatDirectChunkedUploader )
