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

WFCatExportTask.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 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 Disposable = Rx.Disposable
local Observable = Rx.Observable
local ReplaySubject = Rx.ReplaySubject
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

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

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

	-- WF.core.performance
local WFPerformanceReport = require 'WFPerformanceReport'

	-- WF.oz_client.managed_catalog
local WFCatExportQueue = require 'WFCatExportQueue'
local WFCatExportPixelProviderTask = require 'WFCatExportPixelProviderTask'

local NmbProfileDownloaderTask = require 'NmbProfileDownloaderTask'
local NmbDocValueToRenderInfo = require 'NmbDocValueToRenderInfo'

-- export providers
local WFCatExportBiggestRenditionProvider =
		require 'WFCatExportBiggestRenditionProvider'

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

local check = assert( WFArguments.check )

local logger = WFLogger 'WFCatExportTask'
	--:setLevel( 'DEBUG' )

local runUntilDone = WFRxRunUntilDone

local numCPUs = require "WFDeviceInfo".numCPUs() or 1

local kThreadCount = math.max( 1, math.floor( numCPUs / 2 ) )

-- override based on the config setting
local exportConfig = require "WFConfiguration".exportConfig or {}
if exportConfig.numThreads then
	kThreadCount = exportConfig.numThreads
end

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

-- @mark Processing Functions

--------------------------------------------------------------------------------
local _exportRenditionCreator = {
	original = require 'WFCatExportOriginalProvider',
	jpeg = require 'WFCatExportGenericProvider',
	tiff = require 'WFCatExportGenericProvider',
	dng = require 'WFCatExportGenericProvider',
}

--------------------------------------------------------------------------------
--[[!
	\brief Processes one request.
	Steps involved:
		1. Get the original for asset id.
		2. Generate output as required.
		3. Save the output in disk or provide the bytes.
]]--
local function processOneRequest( self, request )

	local assetId = request.id 
	
	if not assetId then
		logger:warnf( 'Export failed as asset id supplied is nil' )
		return Observable.throw( 'invalid_asset_id' )
	end
	
	if not request.settings then
		logger:warnf( 'Export failed as settings supplied is nil' )
		return Observable.throw( 'invalid_settings' )
	end
	
	return self._wfCatalog:docStore():docValue( assetId )
		:map( function( docValue )
			
			if docValue._empty or docValue.type ~= 'asset' then
				logger:warnf( 'Could not find asset with asset id: %s', tostring( assetId ) )
				error( 'invalid_asset_id' )
			end
			
			-- SPAM [ ashj: 2016-03-28 ] Video export is
			-- limited to exporting it as original. 
			-- Once we have proper video format support, 
			-- we might be able to remove this specific handling.

			local availableBinaryStatusForDocValue =
					NmbDocValueToRenderInfo.getAvailableBinaryStatusForDocValue(
						docValue )

			if availableBinaryStatusForDocValue == "no_binaries" then

				if docValue._spaceId then
					error( 'rendition_unavailable' )
				else
					error( 'original_unavailable' )
				end

			end

			if availableBinaryStatusForDocValue == 'full_size_rendition_only' then

				return WFCatExportBiggestRenditionProvider( false )

			end
			
			if availableBinaryStatusForDocValue == '2048_rendition_only' then
				
				return WFCatExportBiggestRenditionProvider( true )
				
			end
			
			local provider
			
			if docValue.subtype == 'video' and docValue._spaceId and availableBinaryStatusForDocValue == "has_proxy" then
				request.settings.plainOriginal = true
				request.settings.export720pRendition = true
				provider = _exportRenditionCreator.original
			elseif docValue.subtype == 'video' then
				request.settings.plainOriginal = true
				provider = _exportRenditionCreator.original
			else
				provider = _exportRenditionCreator[ request.settings.format ]
			end

			-- Special case for heic files, which may choose to bypass or fallback to the default provider
			local fileName = KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' )
			fileName = fileName and string.lower( fileName )
			if fileName and string.match( fileName, '.heic$' ) then
				provider = require 'WFCatExportHeicProvider'( provider )
			end
			
			return provider
			
		end )
		:flatMapLatest( function( provider )
			
			if not provider then
				return Observable.throw( 'invalid_format' )
			end
	
			return WFPerformanceReport.wrapObservable("WFExportTask RenditionCreator.export",
				provider:exportAsset( request, self._wfCatalog, self._exportPixelProviderTask, self._profileDownloaderTask ) )
		end)


end

--------------------------------------------------------------------------------
--[[
	\brief Returns a string which can be used as a
	descriptor for the given asset. 

	SPAM [ ashj: 2016-04-06 ] Though id is used
	to identify the asset within the app, it cannot 
	be used to describe the asset to the user.
	Hence filename is the one used as a descriptor in
	this implementation.
]]

local function getAssetDescriptor( self, assetId )
	
	if not assetId then
		return Observable.just( nil )
	end
	
	return self._wfCatalog:docStore():docValue( assetId )
		:map( function( docValue )

			if docValue._empty or docValue.type ~= 'asset' then
				return assetId
			end

			local fileName = KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' )

			return fileName or assetId
		end)
end

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

local function consumeQueue( self, threadIndex )

	if self:isDisposed() then
		logger:infof( "processor disposed, ending consumeQueue loop" )
		return
	end

	local requestQueue = self._requestQueue

	local request = requestQueue:nextRequest()
		-- TODO [scouten 2015-06-13]: When queue becomes persisent, need to
		-- be able to dequeue the item without removing it from the queue in
		-- the event that the app crashes or quits before fully processing
		-- this file path. For now, this is OK.

	if request == nil then
		logger:infof( "nothing left in queue, ending consumeQueue loop" )
		
		self._isBusy[ threadIndex ]:onNext( false )
		self._isBusy[ threadIndex ]:onCompleted()
		self._isBusy[ threadIndex ]:dispose()
		self._isBusy[ threadIndex ] = nil
		
		-- if all threads are completed, send progress as done
		if next( self._isBusy ) == nil then
			self._progress:done()
		end
		
		return
	end
	
	--ashj-- Instead of using non-main thread for processing
	-- here, let the exportFormatProvider take care of the task
	-- handling.
	Observable.pcall( processOneRequest, self, request )
		:tap(
			function( exportInfo ) -- onNext
				logger:infof( "Exported assetId %s", request.id )
				
				-- If the export succeeded, but with errors/warning, 
				-- we need to treat it as error and send the
				-- extra info.
				if exportInfo.warning then
				
					getAssetDescriptor( self, request.id )
					:tap( function( desc )
						self._progress:reportSuccessMustDeliver( { id = request.id, success = false, warning = exportInfo.warning, data = exportInfo, desc = desc } )
						consumeQueue( self, threadIndex )
					end )
					:call( runUntilDone )
					
				else
					self._progress:reportSuccessMustDeliver( { id = request.id, success = true, data = exportInfo } )
					consumeQueue( self, threadIndex )
				end

			end ,
			function( err )
				logger:errorf( "failed to export assetId %s because %s",
					request.id, err )

				-- ashj: Error is being thrown as a simple lua error
				-- and hence the string will be prefixed by the line
				-- number and file information. Hence taking out only
				-- the last word which corresponds to the error code.
				-- The error thrown has to be a simple word which contains
				-- only _ or any alphanumeric character.
				err = assert( err and err:match( '[_%w]+$' ) )


				-- SPAM [ ashj: 2016-04-06 ] When the export request
				-- fails for an asset, it is required to send an
				-- asset descriptor as well, so that user will be
				-- able to identify the asset.

				getAssetDescriptor( self, request.id )
					:tap( function( desc )
						self._progress:reportErrorMustDeliver( { id = request.id , success = false, error = err, desc = desc } )
						consumeQueue( self, threadIndex )
					end )
					:call( runUntilDone )

			end,
			nil
		)
		:call( runUntilDone )
		
end


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

-- @mark WFCatExportTask class

--------------------------------------------------------------------------------
--[[!
	\brief WFCatExportTaskProcessor gets the assets from the catalog
	and exports them in the required format
]]--

local WFCatExportTask = {

	Disposable,
		
	_wfCatalog = false,
		-- (WFCatalog)

	_requestQueue = false,
		-- (WFCatExportQueue)

	_isBusy = false,
		-- Array of (Rx.ReplaySubject< boolean >) true if processing files

	_progress = false,
		-- (WFProgressReporter) used for export progress updates

	_watchingQueueHandle = false,
		-- (Rx.Disposable) disposal handle for queue watcher
	
	_exportPixelProviderTask = false,

	_profileDownloaderTask = false,
}

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

function WFCatExportTask:__init( wfCatalog, pixelProviderProcessorStream, maybeStyleSynchronizer )

	check( 'WFCatExportTask',
		   'WFCatalog' )
	
	self._wfCatalog = wfCatalog

	self._requestQueue = WFCatExportQueue()
	self._progress = WFProgressReporter( "Exporting files" )
	self._exportPixelProviderTask = WFCatExportPixelProviderTask( pixelProviderProcessorStream )
	if maybeStyleSynchronizer then
		self._profileDownloaderTask = NmbProfileDownloaderTask( wfCatalog, maybeStyleSynchronizer )
	end

	self._requestQueue:totalSize()
		:tap(
			function( value )
				if type( value ) == 'number' then
					self._progress:setTotal( value )
				end
			end
		)
		:call( runUntilDone )
	
	local queueHasContent = self._requestQueue:size()
		:map( function( value )
			return type( value ) == 'number' and value > 0
		end )

	self._isBusy = {}
	
	if type( kThreadCount ) ~= "number" or kThreadCount <= 0 then
		kThreadCount = 1
	end
	
	logger:infof( "WFCatExportTask : No. of threads used: %d", kThreadCount )
	
	for index = 1, kThreadCount do
	
		self._isBusy[ index ] = ReplaySubject( 1 )
		self._isBusy[ index ]:onNext( false )
		
	end
	
	local shouldStartProcessing = queueHasContent
		:withLatestFrom( function( _queueHasContent, _isBusy )
			return _queueHasContent and not _isBusy
		end, self._isBusy[ 1 ] )
		:distinctUntilChanged()
		:filter( function( value )
			return value
				-- Squelch transition from true to false.
		end )

	self._watchingQueueHandle = shouldStartProcessing
		:subscribe( function()
			logger:info( "queue has new content, starting loop" )
			
			for index, isBusyThread in pairs( self._isBusy or {} ) do
				
				isBusyThread:onNext( true )
				consumeQueue( self, index )
			
			end
			
		end )

end

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

function WFCatExportTask:dispose()

	check( 'WFCatExportTask' )

	for index, isBusyThread in pairs( self._isBusy or {} ) do
		
		isBusyThread:onCompleted()
		isBusyThread:dispose()
		self._isBusy[ index ]= nil
		
	end

	if self._watchingQueueHandle then
		self._watchingQueueHandle:dispose()
		self._watchingQueueHandle = nil
	end
	
	if self._exportPixelProviderTask then
		self._exportPixelProviderTask:dispose()
		self._exportPixelProviderTask = nil
	end
	
	if self._profileDownloaderTask then
		self._profileDownloaderTask:close()
	end

	return self:Rx_Disposable_dispose()

end

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

function WFCatExportTask:isBusy()

	check( 'WFCatExportTask' )
	
	local _, busyThread = next( self._isBusy or {} )
	return busyThread or false

end

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

function WFCatExportTask:queueSize()

	check( 'WFCatExportTask' )
	
	return self._requestQueue:size()

end

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

function WFCatExportTask:status()

	check( 'WFCatExportTask' )
	
	return self._progress:status()

end
--------------------------------------------------------------------------------

function WFCatExportTask:exportAssets( requests )

	check( 'WFCatExportTask', 'table' )
	
	if self._progress:isDone() then
		self._progress:reset()
		self._requestQueue:reset()
	end
	
	for _, request in ipairs( requests ) do
		self._requestQueue:addToQueue( request )
	end

end

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

return Rx.Class 'WFCatExportTask'( WFCatExportTask )
