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

WFCatContentRangeUploader.lua

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

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

NOTICE: All information contained herein is, and remains 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 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 Systems Incorporated.

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

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

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

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

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

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

local check = assert( WFArguments.check )
local derefOpt = assert( KSTableUtils.derefOpt )
local KSTableUtils_deepCopy = assert( KSTableUtils.deepCopy )
local logger = WFLogger 'WFCatContentRangeUploader'

local OZ_CHUNK_FILE_SIZE_RECOMMENDED = ( 5 * 1024 * 1024 )
local OZ_CHUNK_FILE_SIZE_REQUIRED = ( 200 * 1024 * 1024 )

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

local function replaceOrInsertHeader( headers, name, value )

	local found = false
	for index, header in ipairs( headers ) do
		if header.header == name then
			if value ~= nil then
				header.value = value
			else
				table.remove( headers, index )
			end
			found = true
			break
		end
	end
	if not found and value ~= nil then
		headers[ #headers + 1 ] = {
			header = name,
			value = value,
		}
	end

end

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

local WFCatContentRangeUploader = {

	_docStore = false,
		-- (WFRxDocStore)

	_catalogSpec = false,
		-- (WFCatSpec)

	_chunkFileSize = WFConfiguration.ChunkedUploader.chunkFileSize or OZ_CHUNK_FILE_SIZE_RECOMMENDED,
		-- (number) Files this size and larger should be uploaded via chunked upload.

	_chunkSize = WFConfiguration.ChunkedUploader.chunkSize or OZ_CHUNK_FILE_SIZE_RECOMMENDED,
		-- (number) Size of chunks used during chunked upload.

	_maxSimultaneous = WFPlatformInfo.isiOS and 100 or 5,
		-- (number) Number of simultaneous chunks to issue. The number relatively set to high for IOS as it uses the background process to upload

	_uploadProgress = false,
		-- (table) set of progressReporter and a variable that tracks total bytes that needs to be uploaded

	_accumulatedChunkBytesSize = 0,
		-- (number) chunk sizes so far accumulated

	WFCloseSupport.addCloseSupportMethods,

}

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

function WFCatContentRangeUploader:__init( docStore, catalogSpec, chunkFileSize, chunkSize, maxSimultaneousTransfers, uploadProgress )

	check( 'WFCatContentRangeUploader', 'WFRxDocStore', 'WFCatSpec', '?number', '?number', '?number' )

	self._docStore = docStore
	self._catalogSpec = catalogSpec
	self._chunkFileSize = math.min( chunkFileSize or self._chunkFileSize, OZ_CHUNK_FILE_SIZE_REQUIRED )
	self._chunkSize = math.min( chunkSize or self._chunkSize, OZ_CHUNK_FILE_SIZE_REQUIRED )
	self._maxSimultaneous = tonumber( maxSimultaneousTransfers ) or self._maxSimultaneous
	if uploadProgress then
		self._uploadProgress = uploadProgress
	end

end

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

function WFCatContentRangeUploader:chunkFileSize()
	return self._chunkFileSize
end

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

function WFCatContentRangeUploader:chunkSize()
	return self._chunkSize
end

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

function WFCatContentRangeUploader:_calculateNextChunk( fileSize, prevEndOffset, existingParts )

	local startOffset = prevEndOffset + 1
	local endOffset = startOffset + self._chunkSize - 1

	if existingParts and #existingParts > 0 then

		for _, part in ipairs( existingParts ) do
			if startOffset >= part.first and startOffset <= part.last then
				startOffset = part.last + 1
				endOffset = startOffset + self._chunkSize - 1 
			end
		end

		for _, part in ipairs( existingParts ) do
			if endOffset >= part.first and endOffset <= part.last and startOffset < part.first then
				endOffset = part.first - 1
			end
		end

		if endOffset - startOffset > self._chunkSize then
			endOffset = startOffset + self._chunkSize - 1
		end

	end

	if startOffset >= fileSize then
		return nil
	end

	if endOffset >= fileSize then
		endOffset = fileSize - 1
	end

	return startOffset, endOffset

end

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

function WFCatContentRangeUploader:_putChunk( url, filePath, putArgs, startOffset, endOffset, fileSize )

	-- 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 contentRangeStr = string.format( "bytes %s-%s/%s", startOffset, endOffset, fileSize)
	replaceOrInsertHeader( putArgs.requestHeaders, 'Content-Range', contentRangeStr )

	-- 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.
	putArgs.dataFormat = 'filePath'
	putArgs.offset = startOffset + 1
	putArgs.length = endOffset - startOffset + 1

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

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

end

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

local function calculateBytesUploaded( existingParts )

	local bytesUploaded = 0
	
	for _, part in ipairs( existingParts ) do
	
		bytesUploaded = bytesUploaded + ( part.last - part.first + 1 )
		
	end
	
	return bytesUploaded

end

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

local function reportProgress( self, existingParts, fileSize, currUploadBytes )

	if self._uploadProgress and self._uploadProgress.progressReporter and fileSize ~= 0 then

		local delta = 0
		
		if existingParts then

			local newUploadBytes = calculateBytesUploaded( existingParts )
			delta = newUploadBytes - currUploadBytes
			currUploadBytes = newUploadBytes

		elseif currUploadBytes > 0 then
			-- this is the case when last chunk got uploaded and it doesn't return existingParts
			delta = fileSize - currUploadBytes 
			currUploadBytes = 0 -- fully uploaded, no history needed. reset to zero.
		end

		if delta > 0 then
			
			self._uploadProgress.totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar + delta
			if( ( self._uploadProgress.totalBytesNeedToUpload - self._uploadProgress.totalBytesUploadedSoFar ) > 0 ) then
				local binaryUploadProgress = {
					totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload,
					totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar,
				}
				self._uploadProgress.progressReporter:updateProgressWithInfo( binaryUploadProgress, true )
			end
			
		end
		
	end

	return currUploadBytes

end

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

function WFCatContentRangeUploader:doChunkedUpload( url, filePath, putArgs, onHttpRequestInProgressCallbackFn )

	check( 'WFCatContentRangeUploader', 'string', 'string', 'table', '?function' )

	local fileSize = WFFileUtils.fileAttributes( filePath ).fileSize or 0
	if fileSize == 0 then
		return Observable.throw( 'payload must exist', {} )
	end

	local putArgsCopy = KSTableUtils_deepCopy( putArgs )

	local requestNextPartGenerator = Rx.ReplaySubject( 1 )

	local function generateNextRequest( fileSize, prevEndOffset, existingParts )

		local nextStart, nextEnd = self:_calculateNextChunk( fileSize, prevEndOffset, existingParts )

		if nextStart then
			requestNextPartGenerator:onNext( nextStart, nextEnd )
		else
			requestNextPartGenerator:onCompleted()
		end

	end

	-- FIXME [jandrew 2021-07-22]: The WFCatContentRangeUploader has two ways to report
	-- progress. One is via _uploadProgress, the newer method via an HTTP-level progressReporter. The
	-- newer method is much more accurate, and at some point we should remove _uploadProgress
	-- altogether. 

	local progressReporter, progressStatusStream

	if putArgs.includeProgress then
		progressReporter = WFProgressReporter( string.format( "doChunkedUpload::%s", url ) )
		progressStatusStream = progressReporter:status()
	end
	
	local currUploadBytes = 0

	-- Put first chunk in order to find out if other parts are already uploaded.
	local initStartOffset, initEndOffset = self:_calculateNextChunk( fileSize, -1, nil )
	
	local requestInProgress = false
	
	local putFirstChunk = self:_putChunk( url, filePath, putArgsCopy, initStartOffset, initEndOffset, fileSize )
	
	if putArgsCopy.notifyWhenHttpRequestIsInProgress then
	
		putFirstChunk = putFirstChunk
	
			:tap( function( _, _, progress )
	
				if not requestInProgress and onHttpRequestInProgressCallbackFn then
					
					if progressReporter then
						progressReporter:setTotal( fileSize, true )
					end
					
					onHttpRequestInProgressCallbackFn()
					requestInProgress = true
					
				end
				
				if progressReporter and progress then
				
					local chunkedProgressInfo = {
						progressOfChunks = { progress }
					}
					
					progressReporter:setProcessedCount( progress.processedCount, true, chunkedProgressInfo )
					
				end
				
			end )
	
			:takeLast( 1 )
		
	end
	
	local chunkedUploadObservable = putFirstChunk
	
		:flatMap( function( responseData, headers )

			local existingParts = nil
			local ok, partsOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )
			
			local bytesUploadedAfterFirstChunk = 0
			
			if ok then
			
				existingParts = derefOpt( partsOrError, 'parts' )
				
				currUploadBytes = reportProgress( self, existingParts, fileSize, currUploadBytes )
				
				if progressReporter then
				
					if existingParts then

						bytesUploadedAfterFirstChunk = calculateBytesUploaded( existingParts )
						progressReporter:setProcessedCount( bytesUploadedAfterFirstChunk, true )
					
					else
					
						local firstChunkSize = initEndOffset - initStartOffset + 1
						bytesUploadedAfterFirstChunk = firstChunkSize
					
					end

				end
				
			else
			
				logger:errorf( "First chunk decode response error %s", partsOrError )
				-- Continue, upload all of the chunks if this happens.
				currUploadBytes = reportProgress( self, {}, 0, 0 )
				
			end

			generateNextRequest( fileSize, initEndOffset, existingParts )

			local chunkedProgressInfo = {
				progressOfChunks = {}
			}

			return requestNextPartGenerator
			
				:map( function( startOffset, endOffset )

					return Observable.just( startOffset, endOffset )
					
						:flatMap( function( startOffset, endOffset )

							generateNextRequest( fileSize, endOffset, existingParts )

							return self:_putChunk( url, filePath, putArgsCopy, startOffset, endOffset, fileSize )
							
							:tap( function( _, _, 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 = bytesUploadedAfterFirstChunk + bytesUploadedForRestOfChunks
									progressReporter:setProcessedCount( bytesUploaded, nil, chunkedProgressInfo )
									
								end
								
							end )
							
							:takeLast( 1 )
							
							:tap( function( responseData, headers )
							
								if self._uploadProgress and self._uploadProgress.progressReporter then
									local ok, partsOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )
									if ok then
										existingParts = derefOpt( partsOrError, 'parts' )
										currUploadBytes = reportProgress( self, existingParts, fileSize, currUploadBytes )
									end
								end
								
							end )

						end )

				end )
			
				:flatMerge( self._maxSimultaneous )
			
				:takeLast( 1 )

		end )
		
	local result = chunkedUploadObservable
	
	if progressStatusStream then
	
		local packedChunkedUploadObservableStream = chunkedUploadObservable
	
			: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 'WFCatContentRangeUploader'( WFCatContentRangeUploader )
