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

PixelProviderDocstoreInfoTask.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 KSTableUtils = require 'KSTableUtils'
local WFFunctionUtils = require "WFFunctionUtils"
local WFPerformanceUtils = require 'WFPerformanceUtils'

	-- pixel provider
local NmbDocValueToRenderInfoStream = require 'NmbDocValueToRenderInfoStream'
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local NmbLocalRenditionDigest = require 'NmbLocalRenditionDigest'
local WFCatRenderSize = require 'WFCatRenderSize'

	-- Rx
local Rx = require 'Rx'

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

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

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

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

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

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt,
										"KSTableUtils.derefOpt does not exist" )

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

local WFCatRenderSize_getMaxCroppedSide = 
		assert( WFCatRenderSize.getMaxCroppedSide,
					"WFCatRenderSize.getMaxCroppedSide does not exist" )

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

local WFCatRenderSize_getAspectRatio = 
		assert( WFCatRenderSize.getAspectRatio,
					"WFCatRenderSize.getAspectRatio does not exist" )

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

local WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription =
		assert( WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithStaticDescription,
				 "No WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithStaticDescription" )

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

local PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId =
         assert( PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId,
                      "PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId missing!" )

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

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

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

local DocstoreInfoStates = PixelPathProviderStateMachine.DocstoreInfoStates

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

local _needBaseInfoStates = {
	assert( DocstoreInfoStates.NeedDownloadOriginalInfo ),
	assert( DocstoreInfoStates.NeedDownloadProxyInfo ),
	assert( DocstoreInfoStates.NeedBestPossibleLocalNegativePathInfo ),
	assert( DocstoreInfoStates.NeedAnyLocalNegativePathInfo ),
	assert( DocstoreInfoStates.NeedAssetInfoForLoupeRendition ),
	assert( DocstoreInfoStates.NeedAssetInfoForExportFullRendition ),
	assert( DocstoreInfoStates.NeedAssetInfoForExportBiggestPossibleRendition ),
	assert( DocstoreInfoStates.NeedAssetInfoForRender ),
	assert( DocstoreInfoStates.NeedXmpDevelopInfo ),
	assert( DocstoreInfoStates.NeedXmpSidecarInfo ),
}

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

local _needBaseInfoFromDocstoreStates = {
	assert( DocstoreInfoStates.NeedDownloadOriginalInfoFromDocstore ),
	assert( DocstoreInfoStates.NeedDownloadProxyInfoFromDocstore ),
	assert( DocstoreInfoStates.NeedBestPossibleLocalNegativePathInfoFromDocstore ),
	assert( DocstoreInfoStates.NeedAnyLocalNegativePathInfoFromDocstore ),
	assert( DocstoreInfoStates.NeedAssetInfoForLoupeRenditionFromDocstore ),
	assert( DocstoreInfoStates.NeedAssetInfoForExportFullRenditionFromDocstore ),
	assert( DocstoreInfoStates.NeedAssetInfoForExportBiggestPossibleRendition ),
	assert( DocstoreInfoStates.NeedAssetInfoForRenderFromDocstore ),
	assert( DocstoreInfoStates.NeedXmpDevelopInfoFromDocstore ),
	assert( DocstoreInfoStates.NeedXmpSidecarInfoFromDocstore ),
}

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

local _fetchingBaseInfoStates = {
	assert( DocstoreInfoStates.FetchingDownloadOriginalInfo ),
	assert( DocstoreInfoStates.FetchingDownloadProxyInfo ),
	assert( DocstoreInfoStates.FetchingAnyLocalNegativePathInfo ),
	assert( DocstoreInfoStates.FetchingBestPossibleLocalNegativePathInfo ),
	assert( DocstoreInfoStates.FetchingAssetInfoForLoupeRendition ),
	assert( DocstoreInfoStates.FetchingAssetInfoForExportFullRendition ),
	assert( DocstoreInfoStates.FetchingAssetInfoForExportBiggestPossibleRendition ),
	assert( DocstoreInfoStates.FetchingAssetInfoForRender ),
	assert( DocstoreInfoStates.FetchingXmpDevelopInfo ),
	assert( DocstoreInfoStates.FetchingXmpSidecarInfo ),
}

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

local _needRenderSettingsStates = {
	assert( DocstoreInfoStates.NeedCheckRenderSettingsForLoupeRendition ),
	assert( DocstoreInfoStates.NeedCheckRenderSettingsForExportFullRendition ),
	assert( DocstoreInfoStates.NeedCheckRenderSettings ),
	assert( DocstoreInfoStates.NeedCheckRenderSettingsForXmpDevelopInfo ),
	assert( DocstoreInfoStates.UseDefaultRenderSettingsForMissingXmpDevelopRender ),
}

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

local function appendT( ... )

	local appendedT = {}

	local ts = {}

	for i = 1, select( '#', ... ) do

		ts[ i ] = select( i, ... )

		assert( type( ts[ i ] ) == "table",
					string.format( "arg %s is not a table!", i ) )

	end

	local j = 0

	for _, t in ipairs( ts ) do

		for _, v in ipairs( t ) do
		
			j = j + 1

			appendedT[ j ] = v

		end

	end

	return appendedT

end

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

local _needOrFetchingDocstoreInfoStates = 
		appendT( _needBaseInfoStates,
				_needBaseInfoFromDocstoreStates,
					_fetchingBaseInfoStates )

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

local function extractParamsFromRenderInfo( renderInfo )

	local assetId = KSTableUtils_derefOpt( renderInfo, "_id" )

	local localAssetId = KSTableUtils_derefOpt( renderInfo, 
													"_localId" )

	local spaceId = KSTableUtils_derefOpt( renderInfo, "_spaceId" )
	
	local subtype = KSTableUtils_derefOpt( renderInfo, 
													"subtype" )

	local revId = KSTableUtils_derefOpt( renderInfo, 
													"_revId" )

	if not( assetId and localAssetId and revId ) then

		logger:warnf( "Got docStoreValue of type '%s' without needed ids",
						type( renderInfo ) )

		return

	end

	local developStanza = KSTableUtils_derefOpt( renderInfo, "develop" )

	if type( developStanza ) ~= "table" then

		logger:warnf( "Asset id %s has renderInfo with type( renderInfo.develop ) set to '%s'. Cannot use render info.",
						localAssetId, type( developStanza ) )

		return

	end

	local changeCounter = KSTableUtils_derefOpt( renderInfo, 
												"_changeCounter" )

	-- TODO 2015.10.30 
	-- This is where we set the digest. This should be done differently. We
	-- should have ACR generate it for us instead of MD5 on a string.
	local digest, xmpCameraRaw, userOrientation =
			NmbLocalRenditionDigest.computeFromRenderInfo( 
				renderInfo )

	local croppedLongSide = WFCatRenderSize_getMaxCroppedSide( renderInfo )

	local aspectRatio = WFCatRenderSize_getAspectRatio( renderInfo )

	local originalPath = KSTableUtils_derefOpt( renderInfo, "originalPath" )

	local proxyPath = KSTableUtils_derefOpt( renderInfo, "proxyPath" )
	
	local xmpDevelopPath = KSTableUtils_derefOpt( renderInfo, "pathToXmpDevelop" )
	
	local xmpDevelopPathSha256 = KSTableUtils_derefOpt( renderInfo, "pathToXmpDevelopSha256" )
	
	local xmpSidecarPath = KSTableUtils_derefOpt( renderInfo, "pathToXmpSidecar" )

	local xmpSidecarExists = KSTableUtils_derefOpt( renderInfo, "xmpSidecarExists" )

	local renditions = KSTableUtils_derefOpt( renderInfo, "renditions" )

	local invalid_renditions = KSTableUtils_derefOpt( renderInfo, "invalid_renditions" )

	local createFullRenditionURITemplate = KSTableUtils_derefOpt( renderInfo, "createFullRenditionURITemplate" )

	local downloadFullRenditionURI = KSTableUtils_derefOpt( renderInfo, "downloadFullRenditionURI" )

	local proxies = KSTableUtils_derefOpt( renderInfo, "proxies" )
	
	local master = KSTableUtils_derefOpt( renderInfo, "master" )

	local ozMasterLink = KSTableUtils_derefOpt( renderInfo, "ozMasterLink" )
	
	local ozProxyLink = KSTableUtils_derefOpt( renderInfo, "ozProxyLink" )
	
	local ozXmpDevelopLink = KSTableUtils_derefOpt( renderInfo, "ozXmpDevelopLink" )

	local ozXmpLink = KSTableUtils_derefOpt( renderInfo, "ozXmpLink" )

	local ozProxySize = KSTableUtils_derefOpt( renderInfo, "ozProxySize" )
	
	local cameraProfileDigest = KSTableUtils_derefOpt( renderInfo, "cameraProfileDigest" )

	local originalCannotBeRendered = KSTableUtils_derefOpt( renderInfo, "originalCannotBeRendered" )

	local cameraProfileMissing = KSTableUtils_derefOpt( renderInfo, "cameraProfileMissing" )
	
	local hasStandardProfile = KSTableUtils_derefOpt( renderInfo, "hasStandardProfile" )

	logger:infof( "Got render info for %s (localAssetId %s)", assetId, localAssetId )

	return assetId, localAssetId, spaceId, subtype, revId, changeCounter,
			digest, xmpCameraRaw, userOrientation, croppedLongSide,
			aspectRatio, originalPath, proxyPath, renditions,
			invalid_renditions, createFullRenditionURITemplate,
			downloadFullRenditionURI, proxies, master, ozMasterLink,
			ozProxyLink, ozProxySize, xmpDevelopPath, xmpDevelopPathSha256,
			ozXmpDevelopLink, xmpSidecarPath, xmpSidecarExists, ozXmpLink,
			originalCannotBeRendered, cameraProfileDigest,
			cameraProfileMissing, hasStandardProfile

end

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

local function applyReceivedTransition( request, scheduler )

	scheduler:_applyTransitionToRequest( request, 'received_docstore_read' )

end

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

local function applyNeedXmpDevelopOrAlreadyPresentTransition( request, scheduler, baseRenderInfoCache, currRenderSettingsCache )
	
	local _, localAssetId, _, _, activeState = unpack( request )
	
	local baseRenderInfo = baseRenderInfoCache:peekCacheItem( localAssetId ) or
							{}

	local digest = baseRenderInfo.digest
	local revisionId = baseRenderInfo.revId
	local xmpCameraRaw = baseRenderInfo.xmpCameraRaw
	local userOrientation = baseRenderInfo.userOrientation
	
	if activeState == DocstoreInfoStates.UseDefaultRenderSettingsForMissingXmpDevelopRender then
		-- SPAM [matthewj 2016-11-22]: We need to create a render for an asset with external develop
		-- settings which are not yet available (and may never be). In this case we'll render with 
		-- default settings (can we render with previous settings?)
		--
		-- To do this we need to replace the baseRenderInfo digest (so we notice changes that will 
		-- fix this) and the currRenderSettingsCache so that we know what to render with.
		
		logger:warnf( "external develop settings unavailable, using default render settings for %s", localAssetId )
		local temporaryXmpCameraRaw = '<xmp/>'
		local temporaryDigest = 
				NmbLocalRenditionDigest.computeTemporaryDefaultForOrientation( 
					baseRenderInfo.ozAssetId or localAssetId, 
					temporaryXmpCameraRaw, userOrientation,
					baseRenderInfo.cameraProfileDigest,
					baseRenderInfo.cameraProfileMissing,
					not baseRenderInfo.hasStandardProfile )
		
		baseRenderInfo.digest = temporaryDigest
		
		baseRenderInfoCache:pushCacheItem( localAssetId, baseRenderInfo )
		
		currRenderSettingsCache:pushCacheItem( localAssetId,
												{ temporaryDigest, revisionId, 
													temporaryXmpCameraRaw, userOrientation } )
		
		scheduler:_applyTransitionToRequest( request, 'use_default_settings' )
		
		return
		
	end
	
	if type( xmpCameraRaw ) == 'table' then
			
		-- SPAM [jandrew 2018-01-19]: 
		-- Here we add incomplete render settings to the cache. Some of these fields,
		-- notably the digest, are used to figure out if we can download a rendition
		-- from Oz. The XMP-settings field is nil for now, but if it turns out we
		-- need to read/download the external XMP, we'll do so and replace this
		-- entry with a complete one.
		
		currRenderSettingsCache:pushCacheItem( 
								localAssetId, 
								{ digest, revisionId,
									nil,
									userOrientation } ) 
			
		scheduler:_applyTransitionToRequest( request, 'xmp_develop_is_external' )
		
		return
	
	end
	
	currRenderSettingsCache:pushCacheItem( 
								localAssetId, 
								{ digest, revisionId,
									xmpCameraRaw,
									userOrientation } ) 
	
	scheduler:_applyTransitionToRequest( request, 'have_xmp_develop' )
	
end

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

local PixelProviderSharedTaskUtils_mustUseOzForRendersForCachedBaseRenderInfo =
		assert( require "PixelProviderSharedTaskUtils".mustUseOzForRendersForCachedBaseRenderInfo,
				"PixelProviderSharedTaskUtils.mustUseOzForRendersForCachedBaseRenderInfo is missing!" )

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

local function processRenderInfo( renderInfo, renderInfoType, 
									scheduler, 
									baseRenderInfoCache,
									currRenderSettingsCache,
									updatedDevelopSettingsAssetIds )

	-- first check if the render info changed in a way we care about
	local assetId, localAssetId, spaceId, subtype, newRevId, newChangeCounter,
			newDigest, newXmpCameraRaw, newUserOrientation, newCroppedLongSide,
			newAspectRatio, newOriginalPath, newProxyPath, newRenditions,
			new_invalid_renditions, newCreateFullRenditionURITemplate,
			newDownloadFullRenditionURI, newProxies, newMaster,
			newOzOriginalLink, newOzProxyLink, newOzProxySize,
			newXmpDevelopPath, newXmpDevelopPathSha256, newOzXmpDevelopLink,
			newXmpSidecarPath, newXmpSidecarExists, newOzXmpLink,
			originalCannotBeRendered, cameraProfileDigest,
			cameraProfileMissing, hasStandardProfile =
				extractParamsFromRenderInfo( renderInfo )

	if not localAssetId then

		return

	end
	
	local maybeBaseRenderInfo = baseRenderInfoCache:peekCacheItem( localAssetId )

	local waitingForDocstoreInfoRequests = 
			scheduler:_getLocalAssetIdRequestsWithState( 
						localAssetId, 
						unpack( _needOrFetchingDocstoreInfoStates ) )

	if #waitingForDocstoreInfoRequests == 0 and not maybeBaseRenderInfo then

		-- no request waiting for this or it's not an update we care care about
		return

	end
	
	local baseRenderInfo = maybeBaseRenderInfo or {}

	local currChangeCounter = baseRenderInfo.changeCounter
	local currDigest = baseRenderInfo.digest
	local currCroppedLongSide = baseRenderInfo.croppedLongSide
	local currAspectRatio = baseRenderInfo.aspectRatio
	local currOriginalPath = baseRenderInfo.originalPath
	local currProxyPath = baseRenderInfo.proxyPath
	local currXmpDevelopPath = baseRenderInfo.xmpDevelopPath
	local currXmpDevelopPathSha256 = baseRenderInfo.xmpDevelopPathSha256
	local currXmpSidecarPath = baseRenderInfo.xmpSidecarPath
	local currXmpSidecarExists = baseRenderInfo.xmpSidecarExists
	local currOzOriginalLink = baseRenderInfo.ozOriginalLink
	local currOzProxyLink = baseRenderInfo.ozProxyLink
	local currOzXmpDevelopLink = baseRenderInfo.ozXmpDevelopLink
	local currOzXmpLink = baseRenderInfo.ozXmpLink
	local currRenditions = baseRenderInfo.renditions
	local curr_invalid_renditions = baseRenderInfo.invalid_renditions
	local currCreateFullRenditionURITemplate = baseRenderInfo.createFullRenditionURITemplate
	local currDownloadFullRenditionURI = baseRenderInfo.downloadFullRenditionURI
	local currRevId = baseRenderInfo.revId
	
	-- need to consider external xmp properly here to decide if it has changed.
	-- if it has, then we need to ensure we fetch it in order to render...
	-- so if we detect a change, and it's external xmp, we need to send to a 
	-- NeedXmpDevelopForRender state that gets it from file ( eventually download )
	-- and then caches as below
	if newDigest == currDigest and
			newCroppedLongSide == currCroppedLongSide and
			newAspectRatio == currAspectRatio and
			newOriginalPath == currOriginalPath and
			newProxyPath == currProxyPath and
			newXmpDevelopPath == currXmpDevelopPath and
			newXmpDevelopPathSha256 == currXmpDevelopPathSha256 and
			newOzOriginalLink == currOzOriginalLink and
			newOzProxyLink == currOzProxyLink and
			newOzXmpDevelopLink == currOzXmpDevelopLink and
			newOzXmpLink == currOzXmpLink and
			newXmpSidecarPath == currXmpSidecarPath and
			newXmpSidecarExists == currXmpSidecarExists and 
			newCreateFullRenditionURITemplate == currCreateFullRenditionURITemplate and
			newDownloadFullRenditionURI == currDownloadFullRenditionURI and
			#newRenditions == #currRenditions and
			#new_invalid_renditions == #curr_invalid_renditions then

		-- no changes we care about
		return

	end

	newChangeCounter = newChangeCounter or 0

	currChangeCounter = currChangeCounter or -1
	
	if newChangeCounter <= currChangeCounter then
	
		scheduler:_sendTestData( 'gotDocstoreInfo', renderInfoType, localAssetId )

		-- don't clobber more recent renderInfo
		return

	end
	
	local updatedBaseRenderInfo =
				{ ozAssetId = assetId, 
					spaceId = spaceId,
					subtype = subtype,
					changeCounter = newChangeCounter, 
					revId = newRevId,
					digest = newDigest, 
					croppedLongSide = newCroppedLongSide,
					aspectRatio = newAspectRatio,
					originalPath = newOriginalPath, 
					proxyPath = newProxyPath, 
					renditions = newRenditions,
					invalid_renditions = new_invalid_renditions,
					createFullRenditionURITemplate = newCreateFullRenditionURITemplate,
					downloadFullRenditionURI = newDownloadFullRenditionURI,
					proxies = newProxies, 
					master = newMaster,
					ozOriginalLink = newOzOriginalLink,
					ozProxyLink = newOzProxyLink,
					ozProxySize = newOzProxySize,
					xmpDevelopPath = newXmpDevelopPath,
					xmpDevelopPathSha256 = newXmpDevelopPathSha256,
					ozXmpDevelopLink = newOzXmpDevelopLink,
					xmpSidecarPath = newXmpSidecarPath,
					xmpSidecarExists = newXmpSidecarExists,
					ozXmpLink = newOzXmpLink,
					userOrientation = newUserOrientation,
					xmpCameraRaw = newXmpCameraRaw,
					originalCannotBeRendered = originalCannotBeRendered,
					cameraProfileDigest = cameraProfileDigest,
					cameraProfileMissing = cameraProfileMissing,
					hasStandardProfile = hasStandardProfile,
				}

	-- cache the base render info
	if not baseRenderInfoCache:pushCacheItem( 
								localAssetId, updatedBaseRenderInfo ) then

		logger:infof( "abandoning baseRenderInfoCache for assetId %s",
						localAssetId )

		return

	end

	-- clear out any previous render settings while we try to fetch the new 
	-- external settings
	currRenderSettingsCache:popCacheItem( localAssetId )

	if currChangeCounter ~= -1 then

		-- we've already received docstore info, this info is an update.
		--
		-- Figure out what changed and notify all requests that have this asset id

		if newDigest ~= currDigest or
				newCroppedLongSide ~= currCroppedLongSide or
				newAspectRatio ~= currAspectRatio or
				newXmpDevelopPath ~= currXmpDevelopPath or
				newXmpDevelopPathSha256 ~= currXmpDevelopPathSha256 or
				newOzXmpDevelopLink ~= currOzXmpDevelopLink then 

			if newOzXmpDevelopLink == currOzXmpDevelopLink and 
					currXmpDevelopPath == nil and 
					newXmpDevelopPath then
				-- If we downloaded xmp develop settings from Oz, and they haven't changed then it's a noop.
			else
				updatedDevelopSettingsAssetIds[ localAssetId ] = true
			end
			
			PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId( 
				scheduler, localAssetId, 'new_docstore_render_info' )

		end

		if ( newRevId ~= currRevId or
					#newRenditions ~= #currRenditions or
					#new_invalid_renditions ~= #curr_invalid_renditions) and 
				PixelProviderSharedTaskUtils_mustUseOzForRendersForCachedBaseRenderInfo( 
					updatedBaseRenderInfo ) then 
			
			PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId( 
				scheduler, localAssetId, 'new_downloadable_links_for_rendition_only_asset' )

		end

		if newOriginalPath ~= currOriginalPath or
				newProxyPath ~= currProxyPath or
				newOzOriginalLink ~= currOzOriginalLink or
				newOzProxyLink ~= currOzProxyLink then

			PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId( 
				scheduler, localAssetId, 'new_docstore_binary_path_info' )

		end

		if newCreateFullRenditionURITemplate ~= 
					currCreateFullRenditionURITemplate or 
				newDownloadFullRenditionURI ~= currDownloadFullRenditionURI then

			PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId( 
				scheduler, localAssetId, 'updated_docstore_fullsize_rendition_link' )

		end

	else

		WFFunctionUtils_map( applyReceivedTransition, 
								waitingForDocstoreInfoRequests,
									scheduler )

	end
	
	scheduler:_sendTestData( 'gotDocstoreInfo', renderInfoType, localAssetId )

end

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

local function processRenderInfos( scheduler, renderInfoType, renderInfos,
										baseRenderInfoCache,
										currRenderSettingsCache,
										updatedDevelopSettingsAssetIds )

	WFFunctionUtils_map( processRenderInfo,
							renderInfos, renderInfoType, scheduler,
							baseRenderInfoCache, currRenderSettingsCache,
							updatedDevelopSettingsAssetIds )

end

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

processRenderInfos =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests(
		"PixelProviderDocstoreInfoTask.processRenderInfos", processRenderInfos, 1 / 16 )

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

local function applyBadAssetIdTransition( request, scheduler )

	scheduler:_applyTransitionToRequest( request, 'bad_asset_id' )

end

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

local function processFailedLocalAssetId( localAssetId, scheduler )

	local waitingForDocstoreInfoRequests = 
			scheduler:_getLocalAssetIdRequestsWithState( 
						localAssetId, 
						unpack( _fetchingBaseInfoStates ) )

	if #waitingForDocstoreInfoRequests == 0 then

		-- no request waiting for this or it's not an update we care care about
		return

	end

	logger:infof( "applying bad_asset_id transition to %s requests for %s",
					#waitingForDocstoreInfoRequests, localAssetId )

	WFFunctionUtils_map( applyBadAssetIdTransition, 
							waitingForDocstoreInfoRequests,
								scheduler )

end

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

local function processFailedLocalAssetIds( scheduler, failedLocalAssetIds )

	WFFunctionUtils_map( processFailedLocalAssetId,
							failedLocalAssetIds, scheduler )

end

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

processFailedLocalAssetIds =
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
			"PixelProviderDocstoreInfoTask", 1 / 16, "processFailedLocalAssetIds", processFailedLocalAssetIds )

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

local function setupTask( scheduler, docStore, baseRenderInfoCache, 
							currRenderSettingsCache, 
							updatedDevelopSettingsAssetIds )

	-- create a subscription that watches for all docstore updates that affect
	-- any active requests on the scheduler
	local allRequestsLocalAssetIdsStream = Rx.Subject()

	local changedRenderInfoStreamOfStreams = 
			Rx.Observable.combineLatest(
				NmbDocValueToRenderInfoStream.changedRenderInfoStream,
					Rx.Observable.just( docStore ), 
					allRequestsLocalAssetIdsStream )

	local changedRenderInfoStream = 
			changedRenderInfoStreamOfStreams:switch()

	local changedRenderInfoSubscription = 
			changedRenderInfoStream
				:subscribe(
					function( renderInfos )

						local ok, errStr =
								pcall( processRenderInfos, 
										scheduler, "changedRenderInfos",
										renderInfos, baseRenderInfoCache,
										currRenderSettingsCache,
										updatedDevelopSettingsAssetIds )

						if not ok then

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

						end

					end,
					function( errStr )

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

					end )


	-- create a subscription that manages pulling the initial docstore value
	-- for a request. This is done in batches.
	local batchStatus = {}

	local batchedLocalAssetIdsStream = Rx.Subject()

	local batchedRenderInfoStreamOfStreams = 
			Rx.Observable.combineLatest(
				NmbDocValueToRenderInfoStream.renderInfoStream,
					Rx.Observable.just( docStore ), 
					batchedLocalAssetIdsStream )

	local batchedRenderInfoStream = 
			batchedRenderInfoStreamOfStreams:switch()

	-- subscribe to the stream so we update requests when it comes in
	local batchedRenderInfoSubscription =
			batchedRenderInfoStream
				:subscribe(
					function( renderInfos, failedLocalAssetIds )

						batchStatus.waiting = false

						local ok, errStr =
								pcall( processRenderInfos, 
										scheduler, "batch", renderInfos,
										baseRenderInfoCache,
										currRenderSettingsCache,
										updatedDevelopSettingsAssetIds )


						if not ok then

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

						end

						ok, errStr =
							pcall( processFailedLocalAssetIds,
									scheduler, failedLocalAssetIds )

						if not ok then

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

						end

					end,
					function( errStr )

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

					end )

	-- return the stream so we can update it and the subscription so we can
	-- dispose it.
	return {

		allRequestsLocalAssetIdsStream = allRequestsLocalAssetIdsStream,

		batchedLocalAssetIdsStream = batchedLocalAssetIdsStream,

		batchStatus = batchStatus,

		renderInfosStreamSubscription = 
			Rx.CompositeDisposable(
				changedRenderInfoSubscription, batchedRenderInfoSubscription )

	}

end

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

local PixelPathProviderSchedulerPrivateMethods_getActiveLocalAssetIdSetCounter =
	PixelPathProviderSchedulerPrivateMethods.getActiveLocalAssetIdSetCounter

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

local PixelPathProviderSchedulerPrivateMethods_getLocalAssetIds =
		PixelPathProviderSchedulerPrivateMethods.getLocalAssetIds

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

local function maybeUpdateLocalAssetIdsStream( scheduler, taskState )

	-- a new scheduler change counter means there may be new localAssetIds that need
	-- renderInfo.  Therefore we must update the inputs to the
	-- renderInfosStreamSubscription so it emits renderInfo for all of the current
	-- localAssetIds

	local currActiveLocalAssetIdSetCounter = 
			PixelPathProviderSchedulerPrivateMethods_getActiveLocalAssetIdSetCounter(
				scheduler )

	if currActiveLocalAssetIdSetCounter ~= 
			taskState.lastActiveLocalAssetIdSetCounter then

		taskState.lastActiveLocalAssetIdSetCounter =
			currActiveLocalAssetIdSetCounter 

		taskState.allRequestsLocalAssetIdsStream
			:onNext( 
				PixelPathProviderSchedulerPrivateMethods_getLocalAssetIds(
					scheduler ) )

	end

end

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

maybeUpdateLocalAssetIdsStream =
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
			"PixelProviderDocstoreInfoTask", 1 / 16, "maybeUpdateLocalAssetIdsStream",
			maybeUpdateLocalAssetIdsStream )

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

local function applyMustFetchOrAlreadyPresentTransition( 
					request, scheduler, baseRenderInfoCache )

	local _, localAssetId = unpack( request )

	local transition = baseRenderInfoCache:peekCacheItem( localAssetId ) and
							'already_present' or 'must_fetch'

	scheduler:_applyTransitionToRequest( request, transition )

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

local function advanceNeedsBaseInfoStates( scheduler, baseRenderInfoCache )

	-- advance the request state for all requests that are waiting for basic info.
	-- if it is in cache we are done, otherwise we will be queuing a docstore
	-- request below
	scheduler:_mapAllRequestsWithStates( _needBaseInfoStates,
					applyMustFetchOrAlreadyPresentTransition, 
						scheduler, baseRenderInfoCache )

end

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

advanceNeedsBaseInfoStates =
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
			"PixelProviderDocstoreInfoTask", 1 / 16, "advanceNeedsBaseInfoStates",
			advanceNeedsBaseInfoStates )

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

local function advanceNeedsXmpDevelopStates( scheduler, baseRenderInfoCache, 
												currRenderSettingsCache )

	-- advance the request state for all requests that are waiting for render settings.
	-- if it is in the cache we are done
	scheduler:_mapAllRequestsWithStates( _needRenderSettingsStates,
		applyNeedXmpDevelopOrAlreadyPresentTransition,
			scheduler, baseRenderInfoCache, 
			currRenderSettingsCache )

end

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

advanceNeedsXmpDevelopStates =
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
			"PixelProviderDocstoreInfoTask", 1 / 16, 
			"advanceNeedsXmpDevelopStates", advanceNeedsXmpDevelopStates )

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

local function validateAndExtractDistinctLocalAssetIds( 
					validatedLocalAssetIdResults, request, scheduler )

	local pixStreamId, localAssetId = unpack( request )

	if type( localAssetId ) ~= "number" or localAssetId <= 0 or
			localAssetId ~= math.floor( localAssetId ) then

		logger:warnf( "Bad local assetId on pixStream '%s'. The bad id: %s",
						pixStreamId, localAssetId )

		scheduler:_applyTransitionToRequest( request, 'bad_asset_id' )

		return validatedLocalAssetIdResults

	end

	validatedLocalAssetIdResults.seen = validatedLocalAssetIdResults.seen or {}

	if not validatedLocalAssetIdResults.seen[ localAssetId ] then

		validatedLocalAssetIdResults.seen[ localAssetId ] = true

		validatedLocalAssetIdResults.distinctLocalAssetIds = 
			validatedLocalAssetIdResults.distinctLocalAssetIds or {}

		validatedLocalAssetIdResults.distinctLocalAssetIds[ 
			#validatedLocalAssetIdResults.distinctLocalAssetIds + 1 ] = localAssetId

	end

	validatedLocalAssetIdResults.validRequests = 
		validatedLocalAssetIdResults.validRequests or {}

	validatedLocalAssetIdResults.validRequests[ 
		#validatedLocalAssetIdResults.validRequests + 1 ] = request 

	return validatedLocalAssetIdResults

end

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

local function applyStartTransition( request, scheduler )

	scheduler:_applyTransitionToRequest( request, 'start_docstore_read' )
					
end

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

local function processRequests( scheduler, taskState, baseRenderInfoCache,
								currRenderSettingsCache )

	-- check to see if we got new localAssetIds since last time
	maybeUpdateLocalAssetIdsStream( scheduler, taskState )

	advanceNeedsBaseInfoStates( scheduler, baseRenderInfoCache )
	
	advanceNeedsXmpDevelopStates( scheduler, baseRenderInfoCache,
									currRenderSettingsCache )

	-- now request some render info in a batch if we aren't already doing so
	if taskState.batchStatus.waiting then

		return

	end

	-- batch up some renderInfo requests
	
	local requests = 
			PixelPathProviderSchedulerPrivateMethods.getNextNRequestsWithState( 
				scheduler, 32, unpack( _needBaseInfoFromDocstoreStates ) )

	if not next( requests ) then

		return

	end

	local validatedLocalAssetIdResults =
			WFFunctionUtils_reduce( {}, 
				validateAndExtractDistinctLocalAssetIds, 
					requests, scheduler )

	if #validatedLocalAssetIdResults.validRequests == 0 then

		return

	end

	taskState.batchStatus.waiting = true

	WFFunctionUtils_map( 
		applyStartTransition, 
			validatedLocalAssetIdResults.validRequests, scheduler )

	taskState.batchedLocalAssetIdsStream
				:onNext( 
					validatedLocalAssetIdResults.distinctLocalAssetIds or {} )

end

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

local function cleanup( taskState )

	taskState.renderInfosStreamSubscription:dispose()

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup

}

