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

PixelProviderPixelCacheCheckTask.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 WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceReport = require 'WFPerformanceReport'

	-- pixel provider
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local WFCatPixelFileCacheCheck = require 'WFCatPixelFileCacheCheck'
local WFCatPixelPathGenerator = require "WFCatPixelPathGenerator"
local WFCatRenderSize = require 'WFCatRenderSize'

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

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

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

local WFCatRenderSize_getBoundedRenderSize =
		assert( WFCatRenderSize.getBoundedRenderSize,
				"WFCatRenderSize.getBoundedRenderSize not defined" )

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

local PixelProviderSharedTaskUtils_maybeUpdateRequestLongSideForMaximumDownloadableSize =
		assert( PixelProviderSharedTaskUtils.maybeUpdateRequestLongSideForMaximumDownloadableSize,
				"PixelProviderSharedTaskUtils.maybeUpdateRequestLongSideForMaximumDownloadableSize does not exist" )

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

local WFCatPixelPathGenerator_pathForPreviewDigestAndLongEdge =
		assert( WFCatPixelPathGenerator.pathForPreviewDigestAndLongEdge,
					"WFCatPixelPathGenerator.pathForPreviewDigestAndLongEdge not defined" )

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

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

--==============================================================================
-- stamina support
--==============================================================================

local WFPerformanceReport_tapObservableOnNext =
		WFPerformanceReport.tapObservableOnNext

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

local WFFunctionUtils_reduce = WFFunctionUtils.reduce

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

local function generateCheckPathResultTupleStat( stats, checkPathResultTuple )

	local _, _, _, _, _, _, renderSize = unpack( checkPathResultTuple )

	renderSize = renderSize or "unknown size"

	stats[ renderSize ] = ( stats[ renderSize ] or 0 ) + 1

	return stats

end

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

local function emitCacheCheckStats( checkPathResultTuples )

	local stats = { totalPathsChecked = #checkPathResultTuples }

	return WFFunctionUtils_reduce( stats, generateCheckPathResultTupleStat,
									checkPathResultTuples )

end

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

local function wrapInStaminaBenchmarking( cacheCheckResultStream )

	return WFPerformanceReport_tapObservableOnNext( 
				"PixelProviderPixelCacheCheckTask",
				cacheCheckResultStream,
					emitCacheCheckStats )

end

--==============================================================================
-- code that passes paths to worker thread checker
--==============================================================================

local function makeCacheCheckResultStream( taskState, checkPathTuples )

	taskState.isDoingCheck = true

	return wrapInStaminaBenchmarking( 
			taskState.cacheChecker:checkTuples( checkPathTuples ) )

end

--==============================================================================
-- code that processes worker thread path check results
--==============================================================================

local PixelCacheCheckStates = 
	PixelPathProviderStateMachine.PixelCacheCheckStates

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

local function makeActiveRenditionInfo( currRenderSettingsCache, localAssetId,
											renditionPath, renderSize,
											croppedLongSide )

	local currRenderSettings = 
			currRenderSettingsCache:peekCacheItem( localAssetId ) or {}

	local currDigest = currRenderSettings[1]
	local currRevId = currRenderSettings[2]
	local currDevSettings = currRenderSettings[3]
	local currVideoDevSettings = currRenderSettings[4]
	local currUserOrientation = currRenderSettings[5]
	local currentTemporal = currRenderSettings[6]
	local currentVideoVersion = currRenderSettings[7]

	currUserOrientation = currUserOrientation or 1

	if not croppedLongSide then
	
		logger:warnf( "croppedLongSide is missing from localAssetId %s will default to renderSize",
							localAssetId, renderSize )

		croppedLongSide = renderSize

	end

	-- save the render info so we can use it in other steps
	return { renditionPath = renditionPath, 
				digest = currDigest, 
				revId = currRevId, 
				renderSize = renderSize,
				croppedLongSide = croppedLongSide, 
				userOrientation = currUserOrientation, 
				temporal = currentTemporal,
				devSettings = currDevSettings,
				videoDevSettings = currVideoDevSettings,
				videoVersion = currentVideoVersion } 

end

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

local function addActiveRenditionInfo( activeRenditionInfoCache, pixStreamId,
										localAssetId, currRenderSettingsCache,
										renditionPath, renderSize,
										croppedLongSide )

	local activeRenditionInfo =
			makeActiveRenditionInfo( 
				currRenderSettingsCache, localAssetId, renditionPath,
				renderSize, croppedLongSide )

	logger:infof( "adding active rendition info for pixStreamId %s localAssetId %s digest %s path %s" ,
					pixStreamId, localAssetId, activeRenditionInfo.digest,
					activeRenditionInfo.renditionPath )

	activeRenditionInfoCache
		:pushCacheItem( pixStreamId, localAssetId, activeRenditionInfo )

	return activeRenditionInfoCache

end

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

local function hasActiveRequestForPathCheckResult( 
				hasActiveRequest, activeRequest, 
				pathCheckPixStreamId, pathCheckLocalAssetId )

	if hasActiveRequest then

		return hasActiveRequest

	end

	local pixStreamId, localAssetId = unpack( activeRequest )

	return pixStreamId == pathCheckPixStreamId and
				localAssetId == pathCheckLocalAssetId 

end

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

local function processPixelCacheCheckTuple( pathCheckResultTuple, scheduler,
											renditionDiskStatusCache,
											activeRenditionInfoCache, 
											currRenderSettingsCache )

	local pathExistsDataArray, pathCheckPixStreamId, pathCheckLocalAssetId, _,
			renditionDigest, croppedLongSide = 
				unpack( pathCheckResultTuple )

	local waitingForCheckRequests =
			scheduler:_getLocalAssetIdRequestsWithState( 
					pathCheckLocalAssetId, 
					PixelCacheCheckStates.CheckingForLoupeRenditionInCache,
					PixelCacheCheckStates.CheckingForExportFullRenditionInCache,
					PixelCacheCheckStates.CheckingForExportBiggestPossibleRenditionInCache,
					PixelCacheCheckStates.CheckingForPixelPathInCache )

	local requestIsStillActive =
			WFFunctionUtils_reduce( false,
				hasActiveRequestForPathCheckResult,
					waitingForCheckRequests, pathCheckPixStreamId,
					pathCheckLocalAssetId )

	if not requestIsStillActive then

		return

	end

	local currDigest = 
			unpack( currRenderSettingsCache:peekCacheItem( 
												pathCheckLocalAssetId ) or {} )

	if not currDigest then

		-- lost our digest info, start over
		scheduler:_maybeSetNextState( pathCheckPixStreamId,
										pathCheckLocalAssetId, 'need_docstore_info' )

		return 'need_docstore_info'

	elseif currDigest ~= renditionDigest then

		-- checked a stale digest, respond appropriately 
		scheduler:_maybeSetNextState( pathCheckPixStreamId,
										pathCheckLocalAssetId, 'stale_check' )

		return 'stale_check'

	end

	local haveIdealRenditionPath = false

	for i, pathExistsData in ipairs( pathExistsDataArray ) do

		local haveRendition, renditionPath, renderSize = unpack( pathExistsData )

		local currRenditionPathIsIdealForRequest = i == 1

		if currRenditionPathIsIdealForRequest then

			haveIdealRenditionPath = haveRendition

		end

		if haveRendition then

			renditionDiskStatusCache
				:addRendition( pathCheckLocalAssetId, renderSize, currDigest,
								renditionPath )

			PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId( 
				scheduler, pathCheckLocalAssetId, 'renditions_status_change' )

		elseif currRenditionPathIsIdealForRequest then

			-- we will either render or download, so set aside data that will
			-- be used by those tasks

			activeRenditionInfoCache = 
				addActiveRenditionInfo( 
					activeRenditionInfoCache, pathCheckPixStreamId,
					pathCheckLocalAssetId, currRenderSettingsCache,
					renditionPath, renderSize, croppedLongSide )

		else

			renditionDiskStatusCache
				:addNoRendition( pathCheckLocalAssetId, renderSize,
									currDigest )

		end

	end

	local transition = 
		haveIdealRenditionPath and 
			'have_local_rendition' or 'no_local_rendition'

	scheduler:_maybeSetNextState( pathCheckPixStreamId, pathCheckLocalAssetId,
									transition )
		
	scheduler:_sendTestData( 'didPixelCacheCheck', pathCheckPixStreamId,
								pathCheckLocalAssetId, transition )

	return transition

end

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

local function updateRequestsWithPathCheckResults( scheduler, 
													pathCheckResultTuples,
													renditionDiskStatusCache,
													activeRenditionInfoCache,
													currRenderSettingsCache )

	return renditionDiskStatusCache:runInBatchMode(
			WFFunctionUtils_map, 
				processPixelCacheCheckTuple, 
					pathCheckResultTuples, scheduler, renditionDiskStatusCache,
					activeRenditionInfoCache, currRenderSettingsCache )

end

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

updateRequestsWithPathCheckResults =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests(
		"PixelProviderDocstoreInfoTask.updateRequestsWithPathCheckResults", 
		updateRequestsWithPathCheckResults, 1 / 12 )

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

local function updateSchedulerWithPathCheckResults( scheduler, taskState,
													renditionDiskStatusCache,
													activeRenditionInfoCache,
													currRenderSettingsCache,
													pathCheckResultTuples )

	taskState.isDoingCheck = false

	return updateRequestsWithPathCheckResults(
			scheduler, pathCheckResultTuples, renditionDiskStatusCache,
			activeRenditionInfoCache, currRenderSettingsCache )

end

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

local function logCacheCheckResults( cacheCheckResults )

	logger:infof( "Finished processing %d cache checks", #cacheCheckResults )

end

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

local function logCacheCheckErrors( errStr )

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

end

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

local function setupTask( scheduler, renditionPathBaseDir,
							renditionDiskStatusCache, activeRenditionInfoCache,
							currRenderSettingsCache )

	local taskState = {}

	taskState.isDoingCheck = false

	-- set up a stream on which we call :onNext when we want to check paths
	taskState.checkPathTuplesStream = Rx.Subject()

	taskState.renditionPathBaseDir = renditionPathBaseDir

	-- set up a worker thread task that checks paths
	taskState.cacheChecker = WFCatPixelFileCacheCheck( renditionPathBaseDir )

	-- set up a stream of results from checking the paths
	local cacheCheckResultStreamOfStreams =
			Rx.Observable.combineLatest(
				makeCacheCheckResultStream,
					Rx.Observable.just( taskState),
					taskState.checkPathTuplesStream )

	local cacheCheckResultStream = 
			cacheCheckResultStreamOfStreams:flatMergeAll()

	-- set up a stream that updates when we've updated the 
	-- scheduler with path check results
	local schedulerUpdatedStream = 
			Rx.Observable.combineLatest(
				updateSchedulerWithPathCheckResults, 
					Rx.Observable.just( scheduler ), 
					Rx.Observable.just( taskState ),
					Rx.Observable.just( renditionDiskStatusCache ),
					Rx.Observable.just( activeRenditionInfoCache ),
					Rx.Observable.just( currRenderSettingsCache ),
					cacheCheckResultStream )

	-- subscribe to the scheduler results stream so that it's active
	taskState.cacheCheckerSubscription = 
			schedulerUpdatedStream 
				:subscribe( logCacheCheckResults, logCacheCheckErrors )

	return taskState

end

--==============================================================================
-- @mark processRequests
--==============================================================================

local function setNextStateAndMaybeGetPathCheckTuple( needsDiskCheckPathTuples, 
											request, scheduler, 
											renditionPathBaseDir,
											renditionDiskStatusCache,
											baseRenderInfoCache,
											currRenderSettingsCache,
											activeRenditionInfoCache )

	local pixStreamId, localAssetId, requestRenderSize, _, requestState = 
			unpack( request )

	local baseRenderInfo, maybeMaxDownloadableRenderSize =
			PixelProviderSharedTaskUtils_maybeUpdateRequestLongSideForMaximumDownloadableSize( 
				scheduler, pixStreamId, localAssetId, requestState,
				baseRenderInfoCache )

	if not baseRenderInfo then

		logger:warnf( "no render info for request: %s.%s.%s",
						pixStreamId, localAssetId, requestState )

		scheduler:_applyTransitionToRequest( request, 'need_docstore_info' )

		return needsDiskCheckPathTuples

	end

	requestRenderSize = maybeMaxDownloadableRenderSize or requestRenderSize

	local assetId, digest, croppedLongSide, subtype = 
			baseRenderInfo.ozAssetId, baseRenderInfo.digest, 
			baseRenderInfo.croppedLongSide, baseRenderInfo.subtype

	local renderSize = WFCatRenderSize_getBoundedRenderSize( 
								croppedLongSide, requestRenderSize, subtype )

	local renditionStatus = renditionDiskStatusCache:getRenditionStatus(
								localAssetId, renderSize, digest )

	if renditionStatus == 0 then

		needsDiskCheckPathTuples[ #needsDiskCheckPathTuples + 1 ] =
			{ pixStreamId, localAssetId, assetId, digest, { renderSize },
				croppedLongSide, requestState, subtype }

		scheduler:_applyTransitionToRequest( request, 'start_rendition_check' )

		return needsDiskCheckPathTuples

	end

	local transition = renditionStatus == 1 and 
							'have_rendition' or 'no_rendition'

	if renditionStatus == -1 then

		-- we will either render or download, so set aside data that will be
		-- used by those tasks

		local renditionPath =
				WFCatPixelPathGenerator_pathForPreviewDigestAndLongEdge(
					renditionPathBaseDir, assetId, digest, renderSize )

		activeRenditionInfoCache = 
			addActiveRenditionInfo( 
				activeRenditionInfoCache, pixStreamId, localAssetId,
				currRenderSettingsCache, renditionPath,
				renderSize, croppedLongSide )

	end

	scheduler:_applyTransitionToRequest( request, transition )

	scheduler:_sendTestData( 'didPixelCacheCheck', pixStreamId, localAssetId,
								transition )

	return needsDiskCheckPathTuples

end

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

local function processRequests( scheduler, pixelCacheCheckTaskState,
									renditionDiskStatusCache,
									baseRenderInfoCache,
									currRenderSettingsCache,
									activeRenditionInfoCache )

	if pixelCacheCheckTaskState.isDoingCheck then
		
		return

	end

	local checkLimit = 32

	local needsDiskCheckPathTuples = {}

	-- loop through all of the requests that need to check if a rendition is
	-- available. Some requests are checked immediately using the
	-- renditionDiskStatusCache. The remaining requests require a async exists
	-- on disk check. Th remaining requests will have a record in
	-- needsDiskCheckPathTuples.
	while #needsDiskCheckPathTuples <= checkLimit do

		local needsCheckRequests =
				PixelPathProviderSchedulerPrivateMethods.getNextNRequestsWithState( 
					scheduler, checkLimit,
					PixelCacheCheckStates.NeedCheckForPixelPathInCache,
					PixelCacheCheckStates.NeedCheckForLoupeRenditionInCache,
					PixelCacheCheckStates.NeedCheckForExportFullRenditionInCache,
					PixelCacheCheckStates.NeedCheckForExportBiggestPossibleRenditionInCache )

		if #needsCheckRequests == 0 then

			break

		end

		needsDiskCheckPathTuples =
			WFFunctionUtils_reduce(
				needsDiskCheckPathTuples,
					setNextStateAndMaybeGetPathCheckTuple,
						needsCheckRequests, 
						scheduler, 
						pixelCacheCheckTaskState.renditionPathBaseDir,
						renditionDiskStatusCache,
						baseRenderInfoCache,
						currRenderSettingsCache,
						activeRenditionInfoCache)

	end

	if #needsDiskCheckPathTuples == 0 then

		return

	end

	-- push them on the stream, which kicks off path checks on all
	pixelCacheCheckTaskState.checkPathTuplesStream
								:onNext( needsDiskCheckPathTuples )

end

--==============================================================================
-- @mark cleanup
--==============================================================================

local function cleanup( pixelCacheCheckTaskState )

	pixelCacheCheckTaskState.cacheCheckerSubscription:dispose()

	pixelCacheCheckTaskState.cacheChecker:close()

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup

}
