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

PixelProviderRenderSaveTask.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 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.

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


	-- wf.core
local KSData = require 'KSData'
local WFFunctionUtils = require 'WFFunctionUtils'

	-- pixel provider
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelProviderSharedBlobCache = 
		require 'PixelProviderSharedBlobCache'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local WFCatPixelFileCacheSave = require 'WFCatPixelFileCacheSave'
local WFFiniteStateMachine = require 'WFFiniteStateMachine'

	-- Rx
local Rx = require 'Rx'
								
--------------------------------------------------------------------------------

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

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

local _1Mb = 1024 * 1024 

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

local _maxBytesInFlight = 256 * _1Mb

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

local WFFunctionUtils_map = assert( WFFunctionUtils.map,
			"WFFunctionUtils.map does not exist" )

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

local WFFunctionUtils_filter = assert( WFFunctionUtils.filter,
			"WFFunctionUtils.filter does not exist" )

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

local WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

--------------------------------------------------------------------------------
-- setupTask
--------------------------------------------------------------------------------

local SaveRenderStates = PixelPathProviderStateMachine.SaveRenderStates

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

local isSavingRenderStates =
		PixelProviderSharedBlobCache.appendT( 
			{ WFFiniteStateMachine.makeRetryableStates(
					PixelPathProviderStateMachine.SaveRenderRetryCount, 
					SaveRenderStates.SavingRender ) },
			{ PixelPathProviderStateMachine.SaveRenderStates.SavingLoupeRendition,
			  PixelPathProviderStateMachine.SaveRenderStates.SavingExportFullRendition,
			  PixelPathProviderStateMachine.SaveRenderStates.SavingBiggestPossibleRenditionDownload } )

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

local fulfilledBySuccessfulRenditionSaveStates = 
		PixelProviderSharedBlobCache.appendT( 
			isSavingRenderStates,
			PixelProviderSharedBlobCache.needRenderDownloadStates )

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

local function sendSaveResultTestData( request, scheduler, msg )

	local pixStreamId, localAssetId, _, _, state = unpack( request )

	scheduler:_sendTestData( msg, pixStreamId, localAssetId, state )

end

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

local PixelProviderSharedBlobCache_getEquivalentRenditionRequests =
			PixelProviderSharedBlobCache.getEquivalentRenditionRequests

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

local PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId =
		PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId

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

local function handleSaveResultImp( scheduler, renditionDiskStatusCache,
									activeRenditionInfoCache,
									pendingSaveBlobCache, ok, maybeErr,
									localAssetId, maybeRenderBlob, renderSize,
									digest, path )

	local fulfilledStates = ok and fulfilledBySuccessfulRenditionSaveStates or
									isSavingRenderStates	

	local fulfilledRequests = 
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache, localAssetId, renderSize,
				digest, unpack( fulfilledStates ) )

	if not next( fulfilledRequests ) then

		logger:warnf( "got an orphaned save result for localAssetId: %s, renderSize: %s, digest: %s, searching { '%s' }",
						localAssetId, renderSize, digest, 
						table.concat( { unpack( fulfilledStates ) }, "','" ) )

		return
						
	end

	if not ok then

		logger:warnf( "error saving localAssetId %s: %s", localAssetId, maybeErr )

		pendingSaveBlobCache:pushBlob( localAssetId, maybeRenderBlob,
										'rendition', renderSize, digest )

	else

		logger:infof( "Did save render blob size %s for localAssetId %s digest %s to path %s" ,
						renderSize, localAssetId, digest, path )

		renditionDiskStatusCache:addRendition( localAssetId, renderSize,
												digest, path )

		PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId(
			scheduler, localAssetId, 'renditions_status_change' )

	end

	local transition = ok and 'saved_render' or 'retry_save_render'

	scheduler:_applyTransitionAllRequests( fulfilledRequests, transition )
		
	WFFunctionUtils_map( sendSaveResultTestData, 
							fulfilledRequests, scheduler,
							ok and 'didPixelCacheSave' or 'saveFailed' )

end

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

handleSaveResultImp =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests(
		"PixelProviderRenderSaveTask.handleSaveResultImp", handleSaveResultImp, 
		1 / 6 )

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

local function recycleSubscription( taskState, slot )

	local maybeSubscription = taskState.activeSubscriptions[ slot ]

	taskState.activeSubscriptions[ slot ] = nil

	taskState.idleSubscriptions[ slot ] = true

	if not maybeSubscription then

		return

	end

	maybeSubscription:dispose()

end

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

local function handleSaveResult( taskState, slot, scheduler,
									renditionDiskStatusCache,
									activeRenditionInfoCache,
									pendingSaveBlobCache, ok, maybeErr,
									localAssetId, maybeRenderBlob, renderSize,
									digest, path )

	recycleSubscription( taskState, slot )

	handleSaveResultImp( 
		scheduler, renditionDiskStatusCache, activeRenditionInfoCache,
		pendingSaveBlobCache, ok, maybeErr, localAssetId, maybeRenderBlob,
		renderSize, digest, path )

end

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

local function setupTask( scheduler, renditionDiskStatusCache,
							activeRenditionInfoCache, pendingSaveBlobCache,
							maxConcurrentSaves )

	local taskState = { maxConcurrentSaves = maxConcurrentSaves, 
							activeSaves = 0,
							activeSubscriptions = {},
							idleSubscriptions = {} }
	
	-- setup subscription bank
	for i = 1, maxConcurrentSaves do

		taskState.idleSubscriptions[ i ] = true

	end

	-- set up the save thread
	taskState.saveToCacheThread = WFCatPixelFileCacheSave( 
								"PixelProviderBlobSaver", 
								maxConcurrentSaves, _maxBytesInFlight )

	-- setup results stream
	taskState.saveResultsStream = Rx.Subject()

	-- subscribe to the saveResultsStream
	taskState.saveResultsSubscription = 
			taskState.saveResultsStream:subscribe(
				function( ok, maybeErr, slot, localAssetId, maybeRenderBlob,
							renderSize, digest, path )

					local okay, maybeError = 
							pcall( handleSaveResult, 
									taskState, slot, scheduler,
									renditionDiskStatusCache,
									activeRenditionInfoCache,
									pendingSaveBlobCache, ok, maybeErr,
									localAssetId, maybeRenderBlob, renderSize,
									digest, path )

					if not okay then

						logger:errorf( "error in handleSaveResult: %s", maybeError )

					end

				end,
				function( errStr )

					logger:errorf( "error in saveResultsSubscription: %s", errStr )

				end )

	return taskState

end

--------------------------------------------------------------------------------
-- processRequests
--------------------------------------------------------------------------------

local function canScheduleSave( taskState )

	return next( taskState.idleSubscriptions ) ~= nil

end

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

local RenderStates = PixelPathProviderStateMachine.RenderStates

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

local needSaveRenderStates = 
		PixelProviderSharedBlobCache.appendT(
			{ WFFiniteStateMachine.makeRetryableStates(
					PixelPathProviderStateMachine.SaveRenderRetryCount, 
					SaveRenderStates.NeedToSaveRender ) },
			{ PixelPathProviderStateMachine.SaveRenderStates.NeedToSaveLoupeRendition,
			  PixelPathProviderStateMachine.SaveRenderStates.NeedToSaveExportFullRendition,
			  PixelPathProviderStateMachine.SaveRenderStates.NeedToSaveExportBiggestPossibleRenditionDownload } )


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

local needsRenderStates =
		{ WFFiniteStateMachine.makeRetryableStates(
				PixelPathProviderStateMachine.RenderRetryCount,
				RenderStates.NeedsRender ) }

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

local needsRenderBlobStates =
		PixelProviderSharedBlobCache.appendT( 
			PixelProviderSharedBlobCache.needRenderDownloadStates,
			needsRenderStates )

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

local fulfilledByPendingSaveStates = 
			PixelProviderSharedBlobCache.appendT( 
				needSaveRenderStates, needsRenderBlobStates )

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

local saveRenderStates = 
		PixelProviderSharedBlobCache.appendT( needSaveRenderStates, 
												isSavingRenderStates )

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

local function hasSaveRenderRequests( localAssetId, scheduler )

	local saveRenderRequests =	
			scheduler:_getLocalAssetIdRequestsWithState( 
						localAssetId,
						unpack( saveRenderStates ) )

	return next( saveRenderRequests ) and localAssetId or nil

end

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

local function hasNoSaveRenderRequests( localAssetId, scheduler )

	return not hasSaveRenderRequests( localAssetId, scheduler ) and
			localAssetId or nil

end

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

local function destroyAllLocalAssetIdRenditions( localAssetId, 
													pendingSaveBlobCache )

	pendingSaveBlobCache:destroyAllLocalAssetIdRenditions( localAssetId )

end

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

local function addActiveRenderSaveInfo( activeRenderSaves, request, 	
											activeRenditionInfoCache )

	local pixStreamId, localAssetId = unpack( request )

	local activeRenditionInfo =
			activeRenditionInfoCache:peekCacheItem( pixStreamId, localAssetId ) 

	if activeRenditionInfo then

		activeRenderSaves[ localAssetId ] = 
			activeRenderSaves[ localAssetId ] or {}

		activeRenderSaves[ localAssetId ][ activeRenditionInfo.digest ] =
			activeRenderSaves[ localAssetId ][ activeRenditionInfo.digest ] or {}

		activeRenderSaves[ localAssetId ][ activeRenditionInfo.digest ][ activeRenditionInfo.renderSize ] = true

	end

	return activeRenderSaves

end

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

local function gatherActiveRenderSaves( activeRenderSaves, localAssetId, 
											scheduler, 
											activeRenditionInfoCache )

	local requestsWithActiveRenderSave =
			scheduler:_getLocalAssetIdRequestsWithState( localAssetId, 
							unpack( isSavingRenderStates ) )

	return WFFunctionUtils_reduce( activeRenderSaves, 
									addActiveRenderSaveInfo,
										requestsWithActiveRenderSave,
										activeRenditionInfoCache )

end

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

local function fulfillPendingRenders( scheduler, localAssetIdsWithBlob, 
											activeRenditionInfoCache )
	
	local activeRenderSaves = WFFunctionUtils_reduce( {}, 
									gatherActiveRenderSaves, 
										localAssetIdsWithBlob, scheduler,
										activeRenditionInfoCache )

	for localAssetId, digestRenders in pairs( activeRenderSaves ) do

		for digest, renderSizes in pairs( digestRenders ) do

			for renderSize in pairs( renderSizes ) do

				local fulfillableDownloadRequests = 
					PixelProviderSharedBlobCache_getEquivalentRenditionRequests(
						scheduler, activeRenditionInfoCache, localAssetId,
						renderSize, digest, unpack( needsRenderBlobStates ) )

					scheduler:_applyTransitionAllRequests( 
								fulfillableDownloadRequests,
								'has_active_rendition_save' )

			end

		end

	end

end

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

local function processRequest( taskState, scheduler, saveResultsStream,
								saveToCacheThread, activeRenditionInfoCache,
								pendingSaveBlobCache, request )

	local pixStreamId, localAssetId = unpack( request )

	local activeRenditionInfo =
			activeRenditionInfoCache:peekCacheItem( 
										pixStreamId, localAssetId ) 

	if not activeRenditionInfo then

		logger:warnf( "Can't save render for %s.%s: no save path", pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipSaveRenditionNoSettings', pixStreamId, localAssetId )

		-- force it to retrieve again
		scheduler:_applyTransitionToRequest( request, 'no_save_path' )

		return

	end

	local renderBlob = 
			pendingSaveBlobCache:popBlob( 
				localAssetId, 'rendition', activeRenditionInfo.renderSize,
				activeRenditionInfo.digest )

	if not renderBlob then

		logger:warnf( "Can't save render for %s.%s: no blob", pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipSaveRenditionNoBlob', pixStreamId, localAssetId )

		-- force it to retrieve again
		scheduler:_applyTransitionToRequest( request, 'no_render_blob' )

		return

	end

	-- now update all needToSaveStates requests that can
	-- also be fulfilled with this blob. In practice, this is almost always
	-- a single request.

	local fulfilledRequests = 
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache, localAssetId,
				activeRenditionInfo.renderSize, 
				activeRenditionInfo.digest, 
				unpack( fulfilledByPendingSaveStates ) )

	-- move all fulfilled to NeedToSaveRender
	scheduler:_applyTransitionAllRequests( 
				fulfilledRequests, 'has_pending_rendition_save' )

	-- then advance all fulfilled to SavingRender
	scheduler:_applyTransitionAllRequests( fulfilledRequests, 'start_save_render' )

	local slot = next( taskState.idleSubscriptions )

	taskState.idleSubscriptions[ slot ] = nil

	-- renderBlobToSave is destroyed by saveToCacheThread, we make a copy 
	-- so that we can retry later if needed.
	local renderBlobToSave = KSData.make( renderBlob )

	local renderSize = activeRenditionInfo.renderSize

	local digest = activeRenditionInfo.digest

	local path = activeRenditionInfo.renditionPath

	logger:infof( "About to save render blob size %s for pixStreamId %s localAssetId %s digest %s to %s" ,
					renderSize, pixStreamId, localAssetId, digest, path )

	local saveToCacheStream =
		saveToCacheThread:savePixels( path, renderBlobToSave )

	taskState.activeSubscriptions[ slot ] =
		saveToCacheStream
			:subscribe(
				function()
					
					logger:warnf( "Unexpected onData event from saveToCacheThread" )

				end,
				function( errStr )

					saveResultsStream:onNext( false, errStr, slot, localAssetId,
												renderBlob, renderSize, 
												digest )

				end,
				function()

					renderBlob:close()

					-- onCompleted is success
					saveResultsStream:onNext( true, nil, slot, localAssetId,
												nil, renderSize, digest, path )

				end  )

end

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

local function processRequests( scheduler, taskState, pendingSaveBlobCache,
									activeRenditionInfoCache )

	-- first purge stale items from the blob cache
	local localAssetIdsWithBlob = pendingSaveBlobCache:localAssetIdsWithBlob()

	local localAssetIdsWithBlobAndNoRequest =
			WFFunctionUtils_filter(
				hasNoSaveRenderRequests,
					localAssetIdsWithBlob, scheduler )

	WFFunctionUtils_map( destroyAllLocalAssetIdRenditions,
							localAssetIdsWithBlobAndNoRequest, 
							pendingSaveBlobCache )

	fulfillPendingRenders( scheduler, localAssetIdsWithBlob, 
								activeRenditionInfoCache )

	local saveToCacheThread, saveResultsStream = 
			taskState.saveToCacheThread, taskState.saveResultsStream

	repeat

		if not canScheduleSave( taskState ) then

			break

		end

		local request =
			{ PixelPathProviderSchedulerPrivateMethods.getNextRequestWithState( 
				scheduler,  unpack( needSaveRenderStates ) ) }

		if not next( request ) then

			break

		end
								
		processRequest( taskState, scheduler, saveResultsStream,
						saveToCacheThread, activeRenditionInfoCache,
						pendingSaveBlobCache, request )

	until false

end

--------------------------------------------------------------------------------
-- cleanup
--------------------------------------------------------------------------------

local function cleanup( taskState )

	for _, subscription in ipairs( taskState.activeSubscriptions ) do

		subscription:dispose()

	end

	taskState.activeSubscriptions = nil

	taskState.idleSubscriptions = nil

	taskState.saveResultsStream:dispose()

	taskState.saveToCacheThread:close()

	taskState.saveToCacheThread = nil

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup,

	privateUnitTestExports = {

		needsRenderBlobStates = needsRenderBlobStates,

	},

}

