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

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'
local WFRxDocValueWatcher = require "WFRxDocValueWatcher"

	-- pixel provider
local NmbDocValueToRenderInfo = require 'NmbDocValueToRenderInfo'
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 NmbDocValueToRenderInfo_maybeExtractRenderInfoFromDocValue =
		assert( NmbDocValueToRenderInfo.maybeExtractRenderInfoFromDocValue,
			"NmbDocValueToRenderInfo.maybeExtractRenderInfoFromDocValue 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 ),
	assert( DocstoreInfoStates.NeedOriginalOrFallbackProxyLocalNegativePathInfo ),
}

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

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 ),
	assert( DocstoreInfoStates.NeedOriginalOrFallbackProxyLocalNegativePathInfo ),
}

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

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.NeedCheckIfXmpDevelopInlineOrExternal ),
	assert( DocstoreInfoStates.UseDefaultRenderSettingsForMissingXmpDevelopRender ),
}

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

local function copyDocValueRenderInfoFieldToBaseRenderInfo( 
				baseRenderInfo, fieldName, docValueRenderInfo )

	baseRenderInfo[ fieldName ] = docValueRenderInfo[ fieldName ]

	return baseRenderInfo

end

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

local function mapDocValueRenderInfoToBaseRenderInfo( docValueRenderInfo )

	-- 20220816 bkruse 
	--
	-- docValueRenderInfo is generated by calling
	--
	-- 	NmbDocValueToRenderInfo.extractRenderInfoFromDocValue( docValue )
	--
	-- docValueRenderInfo is a subset of the docValue, containing only fields
	-- used to manage settings, proxies, originals, and renditions.
	--
	-- In the early days of LrD, I think I thought docValueRenderInfo was "too
	-- big" to store in a cache, so this code generates yet another subset
	-- called baseRenderInfo, which contains a pixel provider specific selection
	-- of fields from docValueRenderInfo.
	--
	-- In retrospect, that might have been a mistake, baseRenderInfo ended up
	-- being largely a slightly modified mirror of docValueRenderInfo. 
	--
	-- Unfortunately, baseRenderInfo objects are used in many files now and 
	-- it's easier to just maintain this method than eliminate baseRenderInfo.
	local newBaseRenderInfo = {}

	local localAssetId = 
			KSTableUtils_derefOpt( docValueRenderInfo, "_localId" )

	local ozAssetId = KSTableUtils_derefOpt( docValueRenderInfo, "_id" )

	local revId = KSTableUtils_derefOpt( docValueRenderInfo, "_revId" )

	if not( ozAssetId and localAssetId and revId ) then

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

		return

	end

	----------------------------------------
	-- non optional baseRenderInfo fields 
	----------------------------------------
	newBaseRenderInfo.ozAssetId = ozAssetId

	newBaseRenderInfo.revId = revId

	local developStanza = KSTableUtils_derefOpt( docValueRenderInfo, "develop" )

	if type( developStanza ) ~= "table" then

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

		return

	end

	----------------------------------------
	-- computed baseRenderInfo fields 
	----------------------------------------
	
	-- 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, maybeVideoAdjustments, userOrientation,
				maybeTemporal =
			NmbLocalRenditionDigest.computeFromRenderInfo( docValueRenderInfo )

	newBaseRenderInfo.digest = digest

	newBaseRenderInfo.xmpCameraRaw = xmpCameraRaw

	newBaseRenderInfo.videoAdjustments = maybeVideoAdjustments
	
	newBaseRenderInfo.videoDevelopVersion = KSTableUtils_derefOpt( docValueRenderInfo, "videoDevelopVersion" )

	newBaseRenderInfo.userOrientation = userOrientation

	newBaseRenderInfo.temporal = maybeTemporal

	newBaseRenderInfo.croppedLongSide = WFCatRenderSize_getMaxCroppedSide( docValueRenderInfo )

	newBaseRenderInfo.aspectRatio = WFCatRenderSize_getAspectRatio( docValueRenderInfo )

	----------------------------------------
	-- optional renamed baseRenderInfo fields 
	----------------------------------------
	newBaseRenderInfo.spaceId = 
		KSTableUtils_derefOpt( docValueRenderInfo, "_spaceId" )

	newBaseRenderInfo.changeCounter = 
		KSTableUtils_derefOpt( docValueRenderInfo, "_changeCounter" )

	newBaseRenderInfo.xmpSidecarPath = 
		KSTableUtils_derefOpt( docValueRenderInfo, "pathToXmpSidecar" )

	----------------------------------------
	-- optional same name baseRenderInfo fields 
	----------------------------------------
	newBaseRenderInfo = 
		WFFunctionUtils_reduce( newBaseRenderInfo,
			copyDocValueRenderInfoFieldToBaseRenderInfo,
				{
					"subtype",
					"isVideo",
					"originalPath",
					"proxyPath",
					"proxyPathOzBinarySubtype",
					"renditions",
					"invalid_renditions",
					"createFullRenditionURITemplate",
					"downloadFullRenditionURI",
					"proxies",
					"ozValidOriginalLink",
					"ozInvalidOriginalLink",
					"ozProxyLink",
					"ozProxySize",
					"ozXmpDevelopLink",
					"localXmpDevelopPathFromDocstore",
					"externalDevelopFiles",
					"xmpSidecarExists",
					"ozXmpLink",
					"originalCannotBeRendered",
					"cameraProfileDigest",
					"cameraProfileMissing",
					"hasStandardProfile",
					"hasDevelopAuxFiles",
					"hasExternalXmpDevelop",
					"hasEmbeddedXmpDevelop",
					"haveAllLocalAuxPathsInDocstore",
				}, docValueRenderInfo )

	return localAssetId, newBaseRenderInfo

end

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

local function isRenderInfoFieldEqual( 
				prevFieldsEqual, fieldName, newRenderInfo, currRenderInfo )

	if not prevFieldsEqual then

		return false

	end

	if not newRenderInfo[ fieldName ] and not currRenderInfo[ fieldName ] then
		
		-- we consider these render info value pairs to be "equal"
		--
		-- new        curr
		-- ---------  ----------
		-- false      false
		-- false      nil
		-- nil        nil
		-- false      nil

		return true

	end

	-- otherwise we use strict equality
	if newRenderInfo[ fieldName ] ~= currRenderInfo[ fieldName ] then

		return false

	end

	return true

end

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

local function areAllRenderInfoFieldsEqual( 
				newRenderInfo, currRenderInfo, fieldNames )

	return WFFunctionUtils.reduce( true, 
			isRenderInfoFieldEqual, 
				fieldNames, newRenderInfo, currRenderInfo )

end

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

local function areRenderInfoArrayFieldCountsEqual( 
				newRenderInfo, currRenderInfo, arrayFieldName )

	if type( newRenderInfo[ arrayFieldName ] ) ~= "table" then

		return false

	end

	if type( currRenderInfo[ arrayFieldName ] ) ~= "table" then

		return false

	end

	return #newRenderInfo[ arrayFieldName ] == 
			#currRenderInfo[ arrayFieldName ]

end

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

local function isRenderSettingRenderInfoChange( newRenderInfo, currRenderInfo )

	local renderInfoFieldsEqual =
			areAllRenderInfoFieldsEqual(
				newRenderInfo, currRenderInfo,
				{
					"digest",
					"croppedLongSide",
					"aspectRatio",
					"ozXmpDevelopLink",
				} )

	local externalXmpFilesAreEqual = 
			renderInfoFieldsEqual and 
			NmbDocValueToRenderInfo.areExternalDevelopFilesEqual(
				newRenderInfo.externalDevelopFiles, 
				currRenderInfo.externalDevelopFiles )

	return not renderInfoFieldsEqual or not externalXmpFilesAreEqual

end

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

local function isLocallyInitiatedRenderSettingsRenderInfoChange( 
				newRenderInfo, baseRenderInfo )

	if not isRenderSettingRenderInfoChange( 
			newRenderInfo, baseRenderInfo ) then

		return false

	end

	if not newRenderInfo.ozXmpDevelopLink and 
			baseRenderInfo.ozXmpDevelopLink then

		-- oz link was removed, almost certainly due to local edit
		return true

	end

	if newRenderInfo.ozXmpDevelopLink == baseRenderInfo.ozXmpDevelopLink and
			newRenderInfo.localXmpDevelopPathFromDocstore and 
			not baseRenderInfo.localXmpDevelopPathFromDocstore then

		-- if we have an external xmp download link and a new local path then
		-- the change is probably that we've synced down the external xmp and
		-- not a local change
		return false

	end

	return true

end

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

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

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

local function isOzOnlyRenderRenditionChange( newRenderInfo, currRenderInfo )
	
	if not PixelProviderSharedTaskUtils_mustUseOzForRendersForCachedBaseRenderInfo( 
			newRenderInfo ) then 

		return false

	end

	if newRenderInfo.revId ~= currRenderInfo.revId then

		return true

	end

	if not areRenderInfoArrayFieldCountsEqual( 
			newRenderInfo, currRenderInfo, "renditions" ) then

		return true

	end

	if not areRenderInfoArrayFieldCountsEqual( 
			newRenderInfo, currRenderInfo, "invalid_renditions" ) then

		return true

	end

	return false

end

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

local function isOzBinaryPathChange( newRenderInfo, currRenderInfo )

	local renderInfoFieldsEqual =
			areAllRenderInfoFieldsEqual(
				newRenderInfo, currRenderInfo,
				{
					"originalPath",
					"proxyPath",
					"ozValidOriginalLink",
					"ozInvalidOriginalLink",
					"ozProxyLink",
				} )

	return not renderInfoFieldsEqual 

end


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

local function isFullSizeRenditionPathChange( newRenderInfo, currRenderInfo )

	local renderInfoFieldsEqual =
			areAllRenderInfoFieldsEqual(
				newRenderInfo, currRenderInfo,
				{
					"createFullRenditionURITemplate",
					"downloadFullRenditionURI",
				} )

	return not renderInfoFieldsEqual 

end

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

local function isXmpSidecarChange( newRenderInfo, currRenderInfo )

	local renderInfoFieldsEqual =
			areAllRenderInfoFieldsEqual(
				newRenderInfo, currRenderInfo,
				{
					"ozXmpLink",
					"xmpSidecarPath",
					"xmpSidecarExists",
				} )

	return not renderInfoFieldsEqual

end

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

local function applyTransitionToRequest( request, scheduler, transition )

	scheduler:_applyTransitionToRequest( request, transition )

end

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

local function handleDefaultSettingsForMissingXmpDevelopRender( 
				scheduler, request, baseRenderInfoCache,
				currRenderSettingsCache, baseRenderInfo )

	local _, localAssetId = unpack( request )

	local userOrientation = baseRenderInfo.userOrientation

	local revisionId = baseRenderInfo.revId

	-- 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 temporaryVideoAdjustmentSettingsDigest = ''

	local temporaryTemporalSettingsDigest = ''

	local temporaryDigest = 
			NmbLocalRenditionDigest.computeTemporaryDefaultForOrientation( 
				baseRenderInfo.ozAssetId or localAssetId, 
				temporaryXmpCameraRaw, 
				temporaryVideoAdjustmentSettingsDigest, 
				baseRenderInfo.userOrientation,
				temporaryTemporalSettingsDigest,
				baseRenderInfo.cameraProfileDigest,
				baseRenderInfo.cameraProfileMissing,
				not baseRenderInfo.hasStandardProfile )
	
	baseRenderInfo.digest = temporaryDigest
	
	baseRenderInfoCache:pushCacheItem( localAssetId, baseRenderInfo )
	
	currRenderSettingsCache
		:pushCacheItem( localAssetId,
			{ 
				temporaryDigest, 
				revisionId, 
				temporaryXmpCameraRaw,
				temporaryVideoAdjustmentSettingsDigest, 
				userOrientation,
				temporaryTemporalSettingsDigest 
			} )
	
	scheduler:_applyTransitionToRequest( request, 'use_default_settings' )

	return

end

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

local function handleVideoDevelopSettings( 
				scheduler, request, currRenderSettingsCache, baseRenderInfo )

	local _, localAssetId = unpack( request )
	
	currRenderSettingsCache
		:pushCacheItem( 
			localAssetId, 
			{ 
				baseRenderInfo.digest, 
				baseRenderInfo.revId,
				nil, -- xmpCameraRaw is not used for videos
				baseRenderInfo.videoAdjustments,
				baseRenderInfo.userOrientation, 
				baseRenderInfo.temporal,
				baseRenderInfo.videoDevelopVersion,
			} ) 
	
	scheduler:_applyTransitionToRequest( request, 'have_xmp_develop' )

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

local function maybeHandleExternalXmpDevelopSettings( 
				scheduler, request, currRenderSettingsCache, baseRenderInfo  )

	local externalSettingsTransition =
			( baseRenderInfo.hasDevelopAuxFiles and
				'xmp_develop_and_aux_files_are_external' ) or
			( baseRenderInfo.hasExternalXmpDevelop and
				'xmp_develop_is_external' ) or
			nil
	
	if not externalSettingsTransition then

		return false

	end
			
	-- 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.
	
	-- SPAM [btran 2022-01-13]
	-- video assets will not end up here because we do not support externalized settings for video.
	
	local _, localAssetId = unpack( request )

	if baseRenderInfo.isVideo then

		logger:warnf( "Video asset %s appears to have unexpected external develop settings which will be ignored", localAssetId )

		return false

	end

	local digest = baseRenderInfo.digest

	local revisionId = baseRenderInfo.revId

	local userOrientation = baseRenderInfo.userOrientation
	
	currRenderSettingsCache
		:pushCacheItem( 
			localAssetId, 
			{ 
				digest, 
				revisionId,
				nil, --xmpCameraRaw
				nil, --videoAdjustmentsSettings
				userOrientation,
				nil, --temporal
				nil, --videoDevelopVersion
			} ) 
		
	scheduler:_applyTransitionToRequest( request, externalSettingsTransition )
	
	return true

end

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

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

	if activeState == 
			DocstoreInfoStates.UseDefaultRenderSettingsForMissingXmpDevelopRender then
		
		handleDefaultSettingsForMissingXmpDevelopRender( 
			scheduler, request, baseRenderInfoCache, currRenderSettingsCache,
			baseRenderInfo )

		return
		
	end

	if baseRenderInfo.isVideo then

		handleVideoDevelopSettings( 
			scheduler, request, currRenderSettingsCache, baseRenderInfo )

		return

	end

	if maybeHandleExternalXmpDevelopSettings( 
			scheduler, request, currRenderSettingsCache, baseRenderInfo  ) then

		return

	end
	
	currRenderSettingsCache
		:pushCacheItem( 
			localAssetId, 
			{ 
				baseRenderInfo.digest, 
				baseRenderInfo.revId, 
				baseRenderInfo.xmpCameraRaw, 
				nil, -- not a video: videoAdjustmentsSetttings is nil
				baseRenderInfo.userOrientation, 
				nil, -- not a video: temporal is nil
				nil, -- not a video: videoDevelopVersion
			} ) 
	
	scheduler:_applyTransitionToRequest( request, 'have_xmp_develop' )
	
end

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

local function maybeAddBaseRenderInfoToCache( 
				baseRenderInfoCache, currRenderSettingsCache, localAssetId,
				baseRenderInfo )

	if not baseRenderInfoCache
			:pushCacheItem( localAssetId, baseRenderInfo ) then

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

		return false

	end

	-- clear out any previous render settings, they will be eventually be
	-- replaced with the baseRenderInfo data
	currRenderSettingsCache:popCacheItem( localAssetId )

	return true

end

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

local _needOrFetchingDocstoreInfoStates

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

local function handleNewRenderInfo( 
				scheduler, baseRenderInfoCache, currRenderSettingsCache,
				renderInfoType, localAssetId, newRenderInfo )

	_needOrFetchingDocstoreInfoStates = 
		_needOrFetchingDocstoreInfoStates or
		scheduler:_getSourceStatesForTransition( 'received_docstore_read' )

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

	-- handle new baseRenderInfo
	if #waitingForDocstoreInfoRequests == 0 then 

		-- no request waiting for this data and it's not cached
		return

	end

	if not maybeAddBaseRenderInfoToCache( 
			baseRenderInfoCache, currRenderSettingsCache, localAssetId,
			newRenderInfo ) then

		return

	end

	WFFunctionUtils_map( applyTransitionToRequest, 
							waitingForDocstoreInfoRequests,
								scheduler, 'received_docstore_read' )

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

end

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

local function isStaleRenderInfoUpdate( newRenderInfo, baseRenderInfo )

	-- make sure the update is newer than our cached render info
	if ( newRenderInfo.changeCounter or 0 ) <= 
			( baseRenderInfo.changeCounter or -1 ) then

		return true

	end

	return false

end

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

local _statesThatRespondToTransitions = {}

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

local function runRenderInfoUpdateResponder( sawRelevantRenderInfoUpdate, 
				renderInfoUpdateResponderInfo, scheduler, baseRenderInfoCache,
				currRenderSettingsCache, localAssetId, newRenderInfo,
				baseRenderInfo )
	
	local renderInfoUpdateIsRelevantFunc, transition =
			unpack( renderInfoUpdateResponderInfo )

	if not renderInfoUpdateIsRelevantFunc( newRenderInfo, baseRenderInfo ) then

		return sawRelevantRenderInfoUpdate

	end

	if not sawRelevantRenderInfoUpdate then

		-- cache the data if it hasn't already been cached
		if not maybeAddBaseRenderInfoToCache( 
				baseRenderInfoCache, currRenderSettingsCache, localAssetId,
				newRenderInfo ) then

			return false

		end

	end

	-- Given the transition associated with renderInfoUpdateIsRelevantFunc,
	-- find all states that would respond to it.
	local statesThatRespondToTransition = 
			_statesThatRespondToTransitions[ transition ] or
			scheduler:_getSourceStatesForTransition( transition )

	_statesThatRespondToTransitions[ transition ] = 
		statesThatRespondToTransition

	-- now find out if we have any requests for those states
	local requestsThatRespondToTransition = 
			scheduler
				:_getLocalAssetIdRequestsWithState( 
					localAssetId, unpack( statesThatRespondToTransition ) )

	-- apply the transition to those requests (if any)
	WFFunctionUtils_map( applyTransitionToRequest, 
							requestsThatRespondToTransition,
								scheduler, transition )

	return true

end

--------------------------------------------------------------------------------
-- _renderInfoUpdateRespondersInfo is a tuple of:
-- 	
-- 		renderInfoChangeClassifierFunc,
-- 		renderInfoChangeStateMachineTransition

local _renderInfoUpdateRespondersInfo = {
	{ 
		isOzOnlyRenderRenditionChange, 
		'new_downloadable_links_for_rendition_only_asset',
	},
	{ 
		isOzBinaryPathChange,
		'new_docstore_binary_path_info',
	},
	{
		isFullSizeRenditionPathChange,
		'updated_docstore_fullsize_rendition_link',
	},
	{
		isRenderSettingRenderInfoChange,
		'new_docstore_render_info'
	},
}

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

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

	local localAssetId, newBaseRenderInfo = 
			mapDocValueRenderInfoToBaseRenderInfo( renderInfo )

	if not localAssetId then

		return

	end

	local maybeBaseRenderInfo = 
			baseRenderInfoCache:peekCacheItem( localAssetId )

	if not maybeBaseRenderInfo then

		-- new render infos handled here
		handleNewRenderInfo( 
			scheduler, baseRenderInfoCache, currRenderSettingsCache,
			renderInfoType, localAssetId, newBaseRenderInfo )

		return

	end

	-- we already have render info, check to see if the updated render info affects
	-- any requests

	local baseRenderInfo = 
			assert( type( maybeBaseRenderInfo ) == "table" and
						maybeBaseRenderInfo, 
					"missing expected cached baseRenderInfo" )

	if isStaleRenderInfoUpdate( newBaseRenderInfo, baseRenderInfo ) then
	
		-- don't clobber more recent renderInfo
		scheduler
			:_sendTestData( 
				'gotDocstoreInfo', renderInfoType, localAssetId )

		return

	end

	if isLocallyInitiatedRenderSettingsRenderInfoChange( 
			newBaseRenderInfo, baseRenderInfo ) then

		-- 20220812 bkruse
		--
		-- TODO trying to look at renderInfo changes and guessing if they are
		-- local or sync based is kludgey and error prone. We should instead be
		-- updating updatedDevelopSettingsAssetIds directly whenever the
		-- client saves develop settings.

		updatedDevelopSettingsAssetIds[ localAssetId ] = true

	end

	-- run each of the responders in _renderInfoUpdateRespondersInfo
	local didHandleRenderInfoUpdate =
		WFFunctionUtils_reduce(
			false,
			runRenderInfoUpdateResponder,
				_renderInfoUpdateRespondersInfo,
				scheduler, baseRenderInfoCache, currRenderSettingsCache,
				localAssetId, newBaseRenderInfo, baseRenderInfo )

	if not didHandleRenderInfoUpdate and
		isXmpSidecarChange( newBaseRenderInfo, baseRenderInfo ) then

		if not maybeAddBaseRenderInfoToCache( 
						baseRenderInfoCache, currRenderSettingsCache,
						localAssetId, newBaseRenderInfo ) then

			return

		end

		logger:warnf( "Got updated xmp sidecar data for asset %s (localAssetId %s), but will not update request",
						newBaseRenderInfo.ozAssetId, localAssetId )

	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 changedRenderInfoSubscription = allRequestsLocalAssetIdsStream
			:call( WFRxDocValueWatcher.make, docStore, "changedRenderInfoStream", true )
			:map( function( docValue )
				local renderInfo = NmbDocValueToRenderInfo_maybeExtractRenderInfoFromDocValue( docValue )
				if renderInfo then
					return { renderInfo }
				else
					return nil
				end
			end )
			:filter( function( renderInfos ) return renderInfos end )
			: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,

	privateUnitTestExport = {

		needBaseInfoFromDocstoreStates = _needBaseInfoFromDocstoreStates,

	}
}

