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

PixelProviderDownloadTask.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2016 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law. Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

	-- WF.core
local KSData = require 'KSData'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceUtils = require 'WFPerformanceUtils'
	
	-- WF.oz_client.managed_catalog.pixel_provider
local PixelProviderSharedBlobCache = require 'PixelProviderSharedBlobCache'
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"

	-- WF.oz_client.managed_catalog.file_management
local WFBinaryRequester = require 'WFBinaryRequester'
local WFCatBinaryDownloader = require 'WFCatBinaryDownloader'

	-- WF.web.http_client
local WFRxHttpClient = require 'WFRxHttpClient'

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

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

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

local PixelPathProviderRequestTypes_getName =
		assert( PixelPathProviderRequestTypes.getName,
				"PixelPathProviderRequestTypes.getName does not exist" )

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

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

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

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

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

local MAX_BINARY_DOWNLOAD_FAILURES_BEFORE_GIVING_UP = 6

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

local function sendTestMessage( request, scheduler, msg )

	local pixStreamId, localAssetId = unpack( request )

	scheduler:_sendTestData( msg, pixStreamId, localAssetId )

end

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

local PixelProviderSharedBlobCache_getEquivalentRenditionRequests =
		PixelProviderSharedBlobCache.getEquivalentRenditionRequests

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

local function getAffectedRenditionDownloadRequests( 
				scheduler, localAssetId, revId, baseRenderInfoCache,
				activeRenditionInfoCache, renderSize, revIdDigestMap )

	if not renderSize then

		return {}

	end
	
	local digestForDownload = revId and revIdDigestMap[ revId ]

	if not digestForDownload then

		local baseRenderInfo = 
				baseRenderInfoCache:peekCacheItem( localAssetId )

		local currRevId = baseRenderInfo and baseRenderInfo.revId or nil

		local currDigest = baseRenderInfo and baseRenderInfo.digestForDownload or nil

		digestForDownload = revId and currRevId == revId and currDigest or nil

	end

	local requestsToUpdate = 
			digestForDownload and 
				PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
					scheduler, activeRenditionInfoCache, localAssetId,
					renderSize, digestForDownload, 
					unpack( PixelProviderSharedBlobCache.downloadingRenditionStates ) ) or
				{}

	return requestsToUpdate, digestForDownload

end


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

local function popDownloadRequestInfo( 
				scheduler, revId, ozAssetId, ozBinaryType, ozBinarySizeName,
				baseRenderInfoCache, activeRenditionInfoCache,
				pendingDownloadInfoCache )
						
	local localAssetId, downloadInfoForBinaryBucketId =
			pendingDownloadInfoCache:removePendingDownloadInfo( 
										ozAssetId, ozBinaryType,
										ozBinarySizeName )

	if not localAssetId then

		return

	end
	
	if ozBinaryType == 'original' then
		
		local requestsToUpdate = PixelProviderSharedBlobCache.getActiveNegativeDownloadRequests( 
						scheduler, localAssetId, ozBinaryType )

		return localAssetId, requestsToUpdate, 
				downloadInfoForBinaryBucketId.maybeOriginalBlobSha256

	end
	
	if ozBinaryType == 'proxy' then
		
		local requestsToUpdate = PixelProviderSharedBlobCache.getActiveNegativeDownloadRequests( 
						scheduler, localAssetId, ozBinaryType )

		return localAssetId, requestsToUpdate

	end
		
	if ozBinaryType == 'rendition' then

		local requestsToUpdate, maybeDigest = 
			getAffectedRenditionDownloadRequests( 
				scheduler, localAssetId, revId, baseRenderInfoCache,
				activeRenditionInfoCache, 
				downloadInfoForBinaryBucketId.renderSize, 
				downloadInfoForBinaryBucketId.revIdDigestMap )

		return localAssetId, requestsToUpdate, maybeDigest

	end
				
	if ozBinaryType == 'xmp_develop' then
	
		local requestsToUpdate = PixelProviderSharedBlobCache.getActiveXmpDevelopDownloadRequests( 
						scheduler, localAssetId, ozBinaryType )
		
		return localAssetId, requestsToUpdate

	end

	if ozBinaryType == 'xmp_sidecar' then

		local requestsToUpdate = PixelProviderSharedBlobCache.getActiveXmpSidecarDownloadRequests(
						scheduler, localAssetId, ozBinaryType )
		
		return localAssetId, requestsToUpdate

	end

end

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

local function tallyBinaryError( errorMessage, headers, _, binaryUrl, taskState )
	
	-- NMB-3111 track error codes for binary downloads that are considered possibly permanent. If we get 
	-- enough of these for a given binary we need to mark it as non-existent.  Oz error codes like 1000 or
	-- HTTP error codes like 404 and 410 should be considered here. Track them - in memory only for now - 
	-- and if we get enough for a particular binary, it needs to get marked.
		
	local shouldRecordError = false
	
	-- get and check headers here
	local httpCode = headers and tonumber( headers.status_code ) or nil
	
	local ozCode

	-- The errors over 500 are usually network related and oz server's won't get
	-- a chance to put it's code in them. (i.e. 502 bad gateway is html not json)
	if httpCode and httpCode < 500 then

		-- Only call decode if it looks like JSON to avoid throwing errors while debugging.
		local maybeJSON = WFRxHttpClient.responseContentTypeIsJson( headers )
		if maybeJSON then

			local decodeOk, maybeJson = pcall( require 'KSJsonDecoder'.decodeWebSafeJson, errorMessage )
			if decodeOk then
				ozCode = type( maybeJson ) == 'table' and tonumber( maybeJson.code )
			end

		end

	end

	if httpCode and ( 400 <= httpCode ) and ( httpCode < 500 ) then
	
		-- For now, consider almost all 4xx errors as something the client should consider as a failure that should
		-- mark the binary as permanently failed except for a few - 401, 403 (this is the forced upgrade response),
		-- 407, 408, and 429 (rate-limiting response)
		shouldRecordError = httpCode ~= 401 and httpCode ~= 403 and httpCode ~= 407 and 
							httpCode ~= 408 and httpCode ~= 429 or 
							( httpCode == 403 and ozCode == 4300 )
	end
	
	-- if the error code wasn't helpful, check whether the JSON response is...
	if not shouldRecordError and ozCode == 1000 then
		shouldRecordError = true
	end
	
	local binaryFailureCounts = taskState.permanentBinaryDownloadFailureCounts
	
	if shouldRecordError then
		local oldCount = binaryFailureCounts[ binaryUrl ] or 0
		binaryFailureCounts[ binaryUrl ] = oldCount + 1
	end
	
	return ( binaryFailureCounts[ binaryUrl ] or 0 ) > MAX_BINARY_DOWNLOAD_FAILURES_BEFORE_GIVING_UP
	
end

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

local function handleFailedDownloadResult( scheduler, errorMessage, revId,
												ozAssetId, ozBinaryType, ozBinarySizeName, 
												headers, linksKey, binaryUrl,
												taskState, baseRenderInfoCache,
												activeRenditionInfoCache, pendingDownloadInfoCache )
	
	local localAssetId, requestsToUpdate = 
			popDownloadRequestInfo( 
				scheduler, revId, ozAssetId, ozBinaryType, ozBinarySizeName,
				baseRenderInfoCache, activeRenditionInfoCache,
				pendingDownloadInfoCache )
	
	
	if not localAssetId or not next( requestsToUpdate ) then
		return
	end

	if errorMessage == 'canceledByNewRequest' then
		return requestsToUpdate, 'reschedule_download'
	end
	
	local considerFailurePermanent = tallyBinaryError( errorMessage, headers, linksKey, binaryUrl, taskState )

	-- Use warnf as this is called from handleDownloadResult / makeBinaryRequestStreamHandler / WFBinaryRequester( WFCatBinaryDownloader )
	-- so the error is already sent to analytics via WFCatBinaryDownloader's call to WFAnalytics.trackEventDownloadFailure.
	logger:warnf( "download request error received for %s(%s) blob with ozAssetId = %s: %s",
						ozBinaryType, ozBinarySizeName, ozAssetId, errorMessage or "invalid blob" )


	-- apply doc store annotation for broken binary at a given revision
	-- make sure this is appropriately revision respecting
	if considerFailurePermanent then
	
		taskState.docStore:applyOverlay( 
			{ 
				_id = ozAssetId,
				_localOnly = { 
					binaryDownloadErrors = {
						[ linksKey ] = {
							[ binaryUrl ] = true
						}
					}
				}
			} )
			:call( require 'WFRxRunUntilDone' )
	end

	local transition = considerFailurePermanent and 
								'cannot_retry_download' or 'retry_download'

	return requestsToUpdate, transition
	
end

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

local PixelProviderSharedTaskUtils_ozRenditionNameToRenderSize =
		assert( PixelProviderSharedTaskUtils.ozRenditionNameToRenderSize,
				"PixelProviderSharedTaskUtils.ozRenditionNameToRenderSize is missing" )

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

local function handleSuccessfulDownloadResult( 
					scheduler, maybeKSDataOrError, revId, ozAssetId,
					ozBinaryType, ozBinarySizeName, baseRenderInfoCache,
					activeRenditionInfoCache, pendingDownloadInfoCache,
					pendingSaveBlobCache )

	local binaryBlob = 
			KSData.isData( maybeKSDataOrError ) and
				maybeKSDataOrError or nil

	local localAssetId, requestsToUpdate, maybeUUID = 
			popDownloadRequestInfo( 
				scheduler, revId, ozAssetId, ozBinaryType, ozBinarySizeName,
				baseRenderInfoCache, activeRenditionInfoCache,
				pendingDownloadInfoCache )

 	if not localAssetId then
		
		logger:warnf( "There is no pending download info for oz asset %s",
						ozAssetId )
						
		if binaryBlob then

			binaryBlob:close()

		end
		
		return 

	end
		
	if not next( requestsToUpdate ) then

		logger:warnf( "we tried a %s(%s) blob download for localAssetId %s, but no requests are waiting for it, will discard. This can happen if we delete a pixel request and the download completes before we have time to cancel it.",
					ozBinaryType, ozBinarySizeName, localAssetId )

		if binaryBlob then

			binaryBlob:close()

		end
		
		return

	end

	if binaryBlob and ozBinaryType == 'original' then

		if maybeUUID then

			local originalBlobSha256 = maybeUUID

			pendingSaveBlobCache
				:pushOriginalBlobWithSha256( 
					localAssetId, binaryBlob, 
					originalBlobSha256 )

		else

			pendingSaveBlobCache
				:pushOriginalBlobWithoutSha256( 
					localAssetId, binaryBlob )

		end

		return requestsToUpdate, 'received_download'

	end

	if binaryBlob and ozBinaryType == 'xmp_develop' then

		pendingSaveBlobCache:pushXmpDevelopBlob( localAssetId, binaryBlob )

		return requestsToUpdate, 'received_download'

	end

	if binaryBlob then

		local maybeDigest = maybeUUID

		pendingSaveBlobCache
			:pushBlob( 
				localAssetId, binaryBlob, ozBinaryType, ozBinarySizeName,
				maybeDigest )
		
		return requestsToUpdate, 'received_download'

	end

	-- NOTE [matthewj 2019-01-02]: this code is inadvertently hit in an edge case described in NMB-10652 where if a 0-byte file is 
	-- uploaded to Oz and LrCC wants to download it, it will download the 0-byte file, but think it received nothing and go into this
	-- retry loop until it errors out. The net effect of this is that for 0-byte master files, LrCC shows a stuck sync
	-- which isn't ideal, but does allow a user to identify them whereas succesfully downloading them would hide this sort of
	-- garbage in the user's catalog.
	logger:warnf( "successful %s(%s) blob download for localAssetId %s, but no actual binary was received", 
					ozBinaryType, ozBinarySizeName, localAssetId )
	
	return requestsToUpdate, 'retry_download'
	
end

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

local function handleDownloadResult( 
				scheduler, okay, maybeKSDataOrError, revId, ozAssetId,
				ozBinaryType, ozBinarySizeName, addlInfo, taskState,
				baseRenderInfoCache, activeRenditionInfoCache,
				pendingSaveBlobCache )
	
	local requestsToUpdate, transition
	
	if okay then
		
		requestsToUpdate, transition = 
			handleSuccessfulDownloadResult( 
				scheduler, maybeKSDataOrError, revId, ozAssetId, ozBinaryType,
				ozBinarySizeName, baseRenderInfoCache,
				activeRenditionInfoCache, taskState.pendingDownloadInfoCache,
				pendingSaveBlobCache )
	
	else
		
		addlInfo = addlInfo or {}

		requestsToUpdate, transition = 
			handleFailedDownloadResult( 
				scheduler, maybeKSDataOrError, revId, ozAssetId, ozBinaryType,
				ozBinarySizeName, addlInfo.headers or {}, addlInfo.linkKey,
				addlInfo.url, taskState, baseRenderInfoCache,
				activeRenditionInfoCache, taskState.pendingDownloadInfoCache )
		
	end
	
								
	if requestsToUpdate and transition then
		
		scheduler:_applyTransitionAllRequests( requestsToUpdate, transition )

		WFFunctionUtils_map( sendTestMessage, 
							requestsToUpdate, scheduler,
							transition )
							
	end
end

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

handleDownloadResult =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests(
		"PixelProviderDownloadTask.handleDownloadResult", handleDownloadResult )

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

local function makeBinaryRequestStreamHandler( scheduler, taskState,
												baseRenderInfoCache,
												activeRenditionInfoCache,
												pendingSaveBlobCache )

	return function( okay, maybeKSDataOrError, revId, ozAssetId,
							ozBinaryType, ozBinarySizeName, addlInfo )
		
		handleDownloadResult( 
			scheduler, okay, maybeKSDataOrError, revId, ozAssetId,
			ozBinaryType, ozBinarySizeName, addlInfo, taskState,
			baseRenderInfoCache, activeRenditionInfoCache, 
			pendingSaveBlobCache )
					
	end

end

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

local function subscribeToBinaryRequesterStream( taskState, binaryRequester )

	return binaryRequester
			:getBinaryRequestStream()
			:subscribe( 

				function( okay, maybeKSDataOrError, revId, ozAssetId,
							ozBinaryType, ozBinarySizeName, addlInfo )

					local ok, maybeError =
							pcall( taskState.binaryRequestStreamHandler, 
										okay, maybeKSDataOrError, revId,
										ozAssetId, ozBinaryType,
										ozBinarySizeName, addlInfo )

					if not ok then

						logger:errorf( "error in binaryRequestStreamHandler for %s: %s", 
											ozAssetId, maybeError )

					end
					
				end, 
				function( errStr )
				
					logger:errorf( "error in renditionDownloadResultsStream: %s", errStr )
				
				end ) 

end

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

local function setupTask( scheduler, wfCatalog, baseRenderInfoCache,
							activeRenditionInfoCache, pendingDownloadInfoCache, 
							pendingSaveBlobCache )

	local binaryRequester = WFBinaryRequester( WFCatBinaryDownloader( wfCatalog ) )

	local taskState = {
			
			docStore = wfCatalog:docStore(),
			
			pendingDownloadInfoCache = pendingDownloadInfoCache,

			binaryRequester = binaryRequester,
			
			permanentBinaryDownloadFailureCounts = {}

	}
	
	-- we make a binaryRequestStreamHandler function here so that we can
	-- pass the handler to a mock binary requester later
	taskState.binaryRequestStreamHandler =
			makeBinaryRequestStreamHandler(
				scheduler, taskState, baseRenderInfoCache,
				activeRenditionInfoCache, pendingSaveBlobCache )

	taskState.binaryRequestStreamHandler =
		WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
			"PixelProviderDownloadTask", 1 / 6, "binaryRequestStreamHandler",
				taskState.binaryRequestStreamHandler )

	taskState.binaryRequestStreamSubscription =
		subscribeToBinaryRequesterStream( 
			taskState, binaryRequester, scheduler, baseRenderInfoCache,
			activeRenditionInfoCache, pendingSaveBlobCache )

	return taskState

end

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

local function removePixStreamRequest( deletedPixProviderRequest,
										pendingDownloadInfoCache )

	local pixStreamId, localAssetId, state = 
			unpack( deletedPixProviderRequest )

	pendingDownloadInfoCache:removePixStreamRequest( 
								pixStreamId, localAssetId, state )

end

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

local function maybeGetOzAssetId( baseRenderInfoCache, localAssetId )

	local baseRenderInfo = 
			baseRenderInfoCache:peekCacheItem( localAssetId )

	return baseRenderInfo and baseRenderInfo.ozAssetId or nil

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

local PixelProviderSharedBlobCache_maybeGetDownloadableRenditionSizeNameForDownloadState =
		PixelProviderSharedBlobCache.maybeGetDownloadableRenditionSizeNameForDownloadState

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

local function maybeGetDownloadableRenditionSizeNameForDownloadState(
					downloadState, baseRenderInfoCache, localAssetId, 
					renderSize )

	local ozRenditionSizeName =
		PixelProviderSharedBlobCache_maybeGetDownloadableRenditionSizeNameForDownloadState(
					downloadState, baseRenderInfoCache, localAssetId, renderSize )

	logger:infof( "maybeGetDownloadableRenditionSizeNameForDownloadState( %s, %s, %s ) returns %s",
					downloadState, localAssetId, renderSize,
					ozRenditionSizeName )

	return ozRenditionSizeName

end

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

local function maybeGetDownloadableRenditionSizeName( 
				downloadState, baseRenderInfoCache, activeRenditionInfoCache,
				pixStreamId, localAssetId )

	local activeRenditionInfo =
			activeRenditionInfoCache:peekCacheItem( pixStreamId, localAssetId ) 

	local renderSize = activeRenditionInfo and activeRenditionInfo.renderSize

	if not renderSize then

		return false, "no renderSize from which to compute URI"

	end

	local ozRenditionSizeName =
		maybeGetDownloadableRenditionSizeNameForDownloadState(
			downloadState, baseRenderInfoCache, localAssetId, renderSize )

	if not ozRenditionSizeName then

		return false, 
				string.format( "no oz rendition size name for renderSize %s", 
								renderSize )

	end

	local renderSizeOfOzRendition = 
			PixelProviderSharedTaskUtils_ozRenditionNameToRenderSize( 
				ozRenditionSizeName )

	if renderSizeOfOzRendition and 
			renderSizeOfOzRendition ~= activeRenditionInfo.renderSize then

		-- if we are choosing to download something other than what we asked
		-- for, update the activeRenditionInfoCache so it reflects this new
		-- size

		activeRenditionInfo.renderSize = renderSizeOfOzRendition

		activeRenditionInfoCache
			:pushCacheItem( pixStreamId, localAssetId, activeRenditionInfo ) 

	end

	return true, ozRenditionSizeName

end

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

local function maybeGetFulfillableRequestsAndDownloadInfoForRenditions( 
				scheduler, downloadState, baseRenderInfoCache,
				activeRenditionInfoCache, pixStreamId, localAssetId )

	local activeRenditionInfo =
			activeRenditionInfoCache:peekCacheItem( pixStreamId, localAssetId ) 

	if not activeRenditionInfo then

		return false, "no activeRenditionInfo"

	end

	if not activeRenditionInfo.digest then

		return false, "no rendition digest"

	end

	local ok, downloadableRenditionSizeNameOrErrStr = 
			maybeGetDownloadableRenditionSizeName( 
				downloadState, baseRenderInfoCache, activeRenditionInfoCache,
				pixStreamId, localAssetId )

	if not ok then

		return false, downloadableRenditionSizeNameOrErrStr

	end

	local downloadableRenditionSizeName = downloadableRenditionSizeNameOrErrStr

	local requests = 
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache, localAssetId,
				activeRenditionInfo.renderSize, activeRenditionInfo.digest, 
				unpack( PixelProviderSharedBlobCache.needRenderDownloadStates ) )

	return true, requests, 'rendition', downloadableRenditionSizeName,
			activeRenditionInfo.renderSize, activeRenditionInfo.revId,
			activeRenditionInfo.digest

end

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

local function getFulfillableRequestsAndDownloadInfoForOriginal( 
				scheduler, localAssetId )

	local ozBinaryType = 'original'

	local requests = 
			PixelProviderSharedBlobCache.getNeedNegativeDownloadRequests( 
				scheduler, localAssetId, ozBinaryType )

	local ozBinarySubtype = 'master'

	return requests, ozBinaryType, ozBinarySubtype
	
end

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

local maybeGetBinarySubtypeForProxy =
		PixelProviderSharedTaskUtils.maybeGetBinarySubtypeForProxy

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

local function maybeGetFulfillableRequestsAndDownloadInfoForProxy(
				scheduler, localAssetId, baseRenderInfoCache, preferredSize )

	local ozBinarySubtype = 
			maybeGetBinarySubtypeForProxy( baseRenderInfoCache, localAssetId, preferredSize )

	if not ozBinarySubtype then

		return false, "no downloadable proxy subtype seen"

	end

	local ozBinaryType = 'proxy'

	local requests = 
			PixelProviderSharedBlobCache.getNeedNegativeDownloadRequests( 
				scheduler, localAssetId, ozBinaryType )

	return true, requests, ozBinaryType, ozBinarySubtype
	
end

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

local function getFulfillableRequestsAndDownloadInfoForXmpDevelop( scheduler, localAssetId )
	
	local ozBinaryType = 'xmp_develop'
	
	local requests = 
			PixelProviderSharedBlobCache.getNeedXmpDevelopDownloadRequests( 
				scheduler, localAssetId, ozBinaryType )
	
	return requests, ozBinaryType, ozBinaryType	

end

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

local function getFulfillableRequestsAndDownloadInfoForXmpSidecar( scheduler, localAssetId )

	local ozBinaryType = 'xmp_sidecar'

	local requests = 
			PixelProviderSharedBlobCache.getNeedXmpSidecarDownloadRequests(
				scheduler, localAssetId, ozBinaryType )

	return requests, ozBinaryType, ozBinaryType	

end

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

local function maybeGetFulfillableRequestsAndDownloadInfo( 
				scheduler, baseRenderInfoCache, activeRenditionInfoCache,
				downloadState, pixStreamId, localAssetId, preferredSize )

	if PixelProviderSharedBlobCache.isNeedsRenditionDownloadState(
											downloadState ) then

		return maybeGetFulfillableRequestsAndDownloadInfoForRenditions( 
				scheduler, downloadState, baseRenderInfoCache,
				activeRenditionInfoCache, pixStreamId, localAssetId )

	end

	if PixelProviderSharedBlobCache.isProxyDownloadState( downloadState ) then
		
		return maybeGetFulfillableRequestsAndDownloadInfoForProxy( 
					scheduler, localAssetId, baseRenderInfoCache, preferredSize )

	end
	
	if PixelProviderSharedBlobCache.isOriginalDownloadState( 
											downloadState ) then

		return true,
				getFulfillableRequestsAndDownloadInfoForOriginal( 
					scheduler, localAssetId )

	end

	if PixelProviderSharedBlobCache.isXmpDevelopDownloadState( 
											downloadState ) then
	
		return true, 
				getFulfillableRequestsAndDownloadInfoForXmpDevelop( 
					scheduler, localAssetId, downloadState, 
					baseRenderInfoCache )
												
	end
	
	if PixelProviderSharedBlobCache.isXmpSidecarDownloadState(
											downloadState ) then

		return true,
				getFulfillableRequestsAndDownloadInfoForXmpSidecar( 
					scheduler, localAssetId, downloadState, 
					baseRenderInfoCache )

	end

	return false, 
			string.format( "bogus state: '%s' passed to maybeGetFulfillableRequestsAndDownloadInfo", 
							downloadState )

end

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

local function canJoinPendingSave( pendingSaveBlobCache, localAssetId, binaryType, 
												ozBinarySizeName, digest )

	if binaryType == 'xmp_develop' then

		return pendingSaveBlobCache:peekXmpDevelopBlob( localAssetId )

	end
	
	return pendingSaveBlobCache:peekBlob( localAssetId, binaryType, 
												ozBinarySizeName, digest )

end

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

local function joinPendingDownload( request, pendingDownloadInfoCache,
										ozAssetId, ozBinaryType,
										ozBinarySubtype )

	local pixStreamId, localAssetId = unpack( request )

	pendingDownloadInfoCache:joinPendingDownload( ozAssetId, ozBinaryType,
													ozBinarySubtype,
													pixStreamId, localAssetId )

end

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

local PixelPathProviderSchedulerPrivateMethods_getDeletedRequestsWithState =
	PixelPathProviderSchedulerPrivateMethods.getDeletedRequestsWithState

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

local function maybeGetDownloadParameters(
				scheduler, baseRenderInfoCache, activeRenditionInfoCache,
				downloadState, pixStreamId, localAssetId, preferredSize )

	local maybeOzAssetId = maybeGetOzAssetId( baseRenderInfoCache, localAssetId )

	if not maybeOzAssetId then

		return false, "couldn't get oz asset id"

	end

	local ozAssetId = maybeOzAssetId

	local ok, fulfillableRequestsOrErrStr, ozBinaryType, ozBinarySubtype, 
			renderSize, revId, digest =
		maybeGetFulfillableRequestsAndDownloadInfo( 
			scheduler, baseRenderInfoCache, activeRenditionInfoCache,
			downloadState, pixStreamId, localAssetId, preferredSize)

	if not ok then

		local failReason = 
			type( fulfillableRequestsOrErrStr ) == "string" and
			fulfillableRequestsOrErrStr or 
			"unexpected maybeGetFulfillableRequestsAndDownloadInfo return value"

		return false, failReason

	end

	return true, fulfillableRequestsOrErrStr, ozAssetId, ozBinaryType,
			ozBinarySubtype, renderSize, revId, digest

end

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

local function processDownloadRequest( scheduler, baseRenderInfoCache,
										activeRenditionInfoCache,
										pendingDownloadInfoCache, 
										pendingSaveBlobCache, binaryRequester,
										request )

	local pixStreamId, localAssetId, preferredSize, currPixelProviderPriority, downloadState =
			unpack( request )

	local ok, fulfillableRequestsOrErrStr, ozAssetId, ozBinaryType,
			ozBinarySubtype, renderSize, revId, digest =
		maybeGetDownloadParameters(
			scheduler, baseRenderInfoCache, activeRenditionInfoCache,
			downloadState, pixStreamId, localAssetId, preferredSize )

	local fulfillableRequests = ok and fulfillableRequestsOrErrStr

	------------------------------------------------------------
	-- 1a) Do we have all the info we need to do a download?
	------------------------------------------------------------

	if not ok or type( fulfillableRequests ) ~= "table" then

		local downloadRequestFailReason =
			not ok and fulfillableRequestsOrErrStr or
			"missing expected fulfillableRequests table"

		logger:warnf( "Skipping download for %s:%s because: %s",
						pixStreamId, localAssetId, downloadRequestFailReason )

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

		scheduler:_applyTransitionToRequest( 
						request, 'missing_download_info' )

		-- continue processing requests now that we've transitioned this one
		-- to missing info
		return true

	end

	------------------------------------------------------------
	-- 1b) (extra sanity check) Do we have all the info we need to do a download?
	------------------------------------------------------------
	
	if not next( fulfillableRequests ) then

		-- 2016.03.18 bkruse In theory, this should never happen. If a
		-- pixStreamId is downloadable, then it should be in pixStreamIds
		-- along with any other pixStreamsId that is also satisfied by this
		-- download. IOW, pixStreamIds should be nil or non-empty.

		logger:warnf( "Skipping download of %s(%s) for %s(%s) because of unexpected failure in maybeGetDownloadInfoAndOzAssetId",
						ozBinaryType, ozBinarySubtype, pixStreamId,
						localAssetId )

		scheduler:_applyTransitionToRequest( request, 'cannot_download' )

		-- continue processing requests now that we've transitioned this one
		-- to cannot download
		return true

	end

	local downloadIsBlockedForRequest, reasonDownloadIsBlockedForRequest = 
			pendingDownloadInfoCache:isDownloadBlocked( 
				scheduler, ozBinaryType, currPixelProviderPriority )

	------------------------------------------------------------
	-- 2) Is downloading blocked for this (or all) request?
	------------------------------------------------------------

	if downloadIsBlockedForRequest then
	
		-- if we can't download anything, we likely will generate hundreds of
		-- messages if there are a bunch of requests backlogged, so use 
		-- logger:info instead of logger:warn here
		logger:infof( "Skipping download for %s:%s because %s reported: %s",
						pixStreamId, localAssetId, downloadIsBlockedForRequest,
						reasonDownloadIsBlockedForRequest )

		local transition = downloadIsBlockedForRequest == 'wf' and 
							'no_internet' or 'volume_cannot_save'

		scheduler:_applyTransitionToRequest( request, transition )

		-- continue processing requests now that we've transitioned this
		-- request to cannot download state
		return true

	end

	------------------------------------------------------------
	-- 3) Can we join a save request for the same binary?
	------------------------------------------------------------

	if canJoinPendingSave( pendingSaveBlobCache, localAssetId, ozBinaryType,
							renderSize, digest ) then
	
		logger:infof( "Skipping download of %s(%s) for %s(%s) because there's a pending save that we can join",
						ozBinaryType, ozBinarySubtype, pixStreamId,
						localAssetId )

		scheduler
			:_applyTransitionToRequest( request, 'have_pending_save_for_blob' )

		-- continue processing requests now that we've transitioned this
		-- request to pending save state
		return true

	end

	------------------------------------------------------------
	-- 4) Is downloading temporarily backlogged?
	------------------------------------------------------------
	
	-- if the cache is in a "can't download" state but we *are* going
	-- to download, that means we will pre-empt a lower priority download. We
	-- send notify the unit tests that is occurring so we can check that we are
	-- preempting as expected
	if not pendingDownloadInfoCache:canDownload() then

		logger:infof( "Will preempt a download and schedule one for %s:%s with priority %s",
						pixStreamId, localAssetId,
						PixelPathProviderRequestTypes_getName( currPixelProviderPriority ) ) 
		
		-- notify test code that we are preempting
		scheduler:_sendTestData( 'reschedulingDownload' )

	end

	if not pendingDownloadInfoCache:canStartOrJoinDownload( 
			scheduler, currPixelProviderPriority, ozAssetId, ozBinaryType,
			ozBinarySubtype ) then

		scheduler:_sendTestData( 'skipDownloadDueToBacklog' )

		logger:debugf( "Will not preempt a download because there is no active download with a lower priority than %s",
						PixelPathProviderRequestTypes_getName( currPixelProviderPriority ) ) 

		-- stop processing requests. The download task is in a back logged
		-- state and can't take any more work at the moment 
		return false

	end

	------------------------------------------------------------
	-- 5) Initiate a download for this request
	------------------------------------------------------------

	local haveActiveDownloadRequest =
			pendingDownloadInfoCache
				:hasEntry( ozAssetId, ozBinaryType, ozBinarySubtype )
										
	local noDownloadReason

	if not haveActiveDownloadRequest then

		haveActiveDownloadRequest, noDownloadReason =
			pendingDownloadInfoCache:startNewDownload( 
				binaryRequester, baseRenderInfoCache,
				ozAssetId, ozBinaryType, ozBinarySubtype, localAssetId,
				currPixelProviderPriority, renderSize, revId, digest )

	end

	if not haveActiveDownloadRequest then

		logger:warnf( "Failed to start download of %s:%s ( %s ) %s.%s with priority %s because: %s",
						ozAssetId, ozBinaryType, ozBinarySubtype,
						pixStreamId, localAssetId,
						PixelPathProviderRequestTypes_getName( currPixelProviderPriority ),
						noDownloadReason ) 

		scheduler:_applyTransitionToRequest( request, 'cannot_download' )

		return true

	end

	logger:infof( "Started download of %s:%s ( %s ) and applied `start` on scheduler for %s with priority %s",
					ozAssetId, ozBinaryType, ozBinarySubtype,
					localAssetId,
					PixelPathProviderRequestTypes_getName( currPixelProviderPriority ) ) 

	-- Move all equivalent requests that aren't in download state to download
	-- state

	WFFunctionUtils_map( joinPendingDownload,
							fulfillableRequests,
							pendingDownloadInfoCache, ozAssetId,
							ozBinaryType, ozBinarySubtype )

	-- transition all download state requests to downloading state
	scheduler:_applyTransitionAllRequests( 
				fulfillableRequests, 'start_download' )

	-- continue processing new requests
	return true

end

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

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

	local pendingDownloadInfoCache = taskState.pendingDownloadInfoCache

	local deletedIsDownloadingRequests =
				PixelPathProviderSchedulerPrivateMethods_getDeletedRequestsWithState(
					scheduler,
					unpack( PixelProviderSharedBlobCache.downloadingBlobStates ) )
	
	WFFunctionUtils_map( removePixStreamRequest, 
							deletedIsDownloadingRequests,
								pendingDownloadInfoCache )
	
	pendingDownloadInfoCache:
		maybeCancelExistingDownloadsIfBadStatus( scheduler )

	local request = 
			pendingDownloadInfoCache
				:getNextDownloadToBlobRequest( scheduler )

	if not request then

		-- there are no requests to process, bail
		return

	end

	processDownloadRequest( 
		scheduler, baseRenderInfoCache, activeRenditionInfoCache,
		pendingDownloadInfoCache, pendingSaveBlobCache,
		taskState.binaryRequester, request )

end

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

local function cleanup( taskState )
	
	if taskState.binaryRequestStreamSubscription then

		taskState.binaryRequestStreamSubscription:dispose()

	end

	if taskState.binaryRequester then

		taskState.binaryRequester:close()

	end
	
end

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

local function injectMockBinaryRequester( taskState, maxDownloads,
												mockBinaryRequester )

	cleanup( taskState )

	taskState.binaryRequester = mockBinaryRequester

	taskState.binaryRequestStreamSubscription =
		subscribeToBinaryRequesterStream( taskState, mockBinaryRequester )

	taskState.pendingDownloadInfoCache:setMaxDownloadCount( maxDownloads )

	taskState.pendingDownloadInfoCache:reset()

	return taskState

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup,

	privateUnitTestExport = {

		injectMockBinaryRequester = injectMockBinaryRequester,

	}

}
