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

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 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.requestCost = 1
	
	optionsCopy.requestHeaders = optionsCopy.requestHeaders or {}
	
	local contentType = maybeContentType( optionsCopy.requestHeaders )
	
	local mainPartsUrl = mainPartsRequestUrl
	mainPartsUrl = mainPartsUrl:replace( '{upload_status}', 'pending' )
	mainPartsUrl = mainPartsUrl:replace( '{file_size}', tostring( fileSize ) )
	mainPartsUrl = mainPartsUrl:replace( '{part_size}', tostring( OZ_CHUNK_FILE_SIZE_RECOMMENDED ) )
	mainPartsUrl = mainPartsUrl:replace( '{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 not headers or headers.status_code ~= '200' then

				return Rx.Observable.throw( "Fail to GET main/parts with HTTP status code %s", headers.status_code )
				
			end

			-- decode urlDataJson into lua tables
			local ok, mainPartsOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, mainPartsResponseJson )
			
			if not ok then
				
				return Rx.Observable.throw( "Fail to decode mainParts response error: %s", mainPartsOrError )
			
			end
			
		
			return Observable.just( mainPartsOrError )
			
		end )
		
		:catch( function( responseData, headers )
			return Observable.throw( responseData, headers, mainPartsRequestUrl )
		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 )

			:map( function( data, headers )
				
				if not headers or headers.status_code ~= '202' then
				
					logger:warnf( "Fail to POST complete_upload request with HTTP status code %s", headers.status_code )
					
					return error( "Fail to POST complete_upload request with HTTP status code %s", headers.status_code )
				
				end

				return 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, options, onHttpRequestInProgressCallbackFn, onHttpRequestSuccessCallBackFn )

	check( 'WFCatDirectChunkedUploader', 'string', 'string', 'table', '?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 pendingParts = KSTableUtils_derefOpt( mainPartsResponse, 'parts' ) or {}
			local completeUrl = KSTableUtils_derefOpt( mainPartsResponse, 'links', 'rels/complete_upload', 'href' ) 
			
			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 )
