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

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.

------------------------------------------------------------------------------]]
	-- Rx
local Rx = require "Rx"

	-- WF.core
local KSData = require 'KSData'
local KSTableUtils = require 'KSTableUtils'
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 WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

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

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt, 
										"KSTableUtils.derefOpt undefined" )

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

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

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

local PixelPathProviderRequestTypes_names =
		assert( PixelPathProviderRequestTypes.names,
				"PixelPathProviderRequestTypes.names does not exist" )

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

local PixelPathProviderRequestTypes_comparePriorities =
		assert( PixelPathProviderRequestTypes.comparePriorities,
				"PixelPathProviderRequestTypes.comparePriorities does not exist" )

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

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

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

local getBinarySubtypeForBinaryType = 
		PixelProviderSharedTaskUtils.getBinarySubtypeForBinaryType

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

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

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

local function logAndRaiseError( str, ... )

	local errStr = select( '#', ... ) > 0 and string.format( str, ... ) or str

	logger:error( errStr )

	error( errStr )

end

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

local MAX_BINARY_DOWNLOAD_FAILURES_BEFORE_GIVING_UP = 6

--------------------------------------------------------------------------------
-- PendingDownloadInfoCache
--------------------------------------------------------------------------------

local PendingDownloadInfoCachePrototype = {

	_ozAssetIdInfoMap = false,

	_activePixStreamRequestInfo = false,

	_maxDownloadCount = false,

}

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

function PendingDownloadInfoCachePrototype:__init( maxDownloadCount )

	self._ozAssetIdInfoMap = {}

	self._activePixStreamRequestInfo = {}

	self._maxDownloadCount = maxDownloadCount

end

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

function PendingDownloadInfoCachePrototype:setMaxDownloadCount( maxDownloadCount )

	self._maxDownloadCount = maxDownloadCount

end

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

function PendingDownloadInfoCachePrototype:reset()

	if not next( self._ozAssetIdInfoMap ) then

		return

	end

	self._ozAssetIdInfoMap = {}

	self._activePixStreamRequestInfo = {}

end

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

local function makeBinaryBucketId( ozBinaryType, ozBinarySizeName )

	return string.format( "%s:%s", ozBinaryType, 
									ozBinarySizeName or 'any_size' )

end

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

local function maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	local ozAssetIdInfoMap = 
			self._ozAssetIdInfoMap[ ozAssetId ] 

	if not ozAssetIdInfoMap then

		return

	end

	local binaryBucketId = makeBinaryBucketId( binaryType, binarySize ) 

	-- 1) pop the active downloadInfoForBinaryBucketId

	local downloadInfoForBinaryBucketId = 
			KSTableUtils_derefOpt( ozAssetIdInfoMap,
									'binaryBucketIdDownloadInfoMap',
									binaryBucketId )

	return ozAssetIdInfoMap, binaryBucketId, downloadInfoForBinaryBucketId

end

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

function PendingDownloadInfoCachePrototype:hasEntry( ozAssetId, binaryType,
																binarySize ) 

	local ozAssetIdInfoMap, binaryBucketId, downloadInfoForBinaryBucketId = 
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	return ozAssetIdInfoMap ~= nil and 
			binaryBucketId ~= nil and 
			downloadInfoForBinaryBucketId ~= nil

end

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

function PendingDownloadInfoCachePrototype:allActiveDownloadRequests( )

	local activeDownloadRequests = {}
	
	for pixStreamId, pixStreamRequests in pairs( self._activePixStreamRequestInfo ) do
		
		for localDocId, downloadRequest in pairs( pixStreamRequests ) do
			
			table.insert( activeDownloadRequests, { pixStreamId, localDocId, unpack( downloadRequest ) } )
		
		end
	
	end
	
	return activeDownloadRequests
end

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

function PendingDownloadInfoCachePrototype:getHttpPriorityIfDownloading( 
												ozAssetId, binaryType,
												binarySize ) 

	local _, _, downloadInfoForBinaryBucketId = 
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	if type( downloadInfoForBinaryBucketId ) ~= "table" then

		return

	end

	return downloadInfoForBinaryBucketId.httpPriority 

end

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

function PendingDownloadInfoCachePrototype:getPixelProviderPriorityIfDownloading( 
												ozAssetId, binaryType,
												binarySize ) 

	local _, _, downloadInfoForBinaryBucketId = 
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	if type( downloadInfoForBinaryBucketId ) ~= "table" then

		return

	end

	return downloadInfoForBinaryBucketId.pixelProviderPriority 

end


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

function PendingDownloadInfoCachePrototype:removePendingDownloadInfo( 
					ozAssetId, binaryType, binarySize ) 

	local ozAssetIdInfoMap, binaryBucketId, downloadInfoForBinaryBucketId = 
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	if not ozAssetIdInfoMap then

		return

	end

	-- 1) pop the active downloadInfoForBinaryBucketId

	ozAssetIdInfoMap.binaryBucketIdDownloadInfoMap[ binaryBucketId ] = nil

	if not downloadInfoForBinaryBucketId then

		return

	end

	-- 2) maybe clear out containers that held the popped data

	-- a) clear out map if no other sized downloads for ozAssetId
	if not next( ozAssetIdInfoMap.binaryBucketIdDownloadInfoMap ) then

		self._ozAssetIdInfoMap[ ozAssetId ] = nil

	end

	-- b) Clear out the associated self._activePixStreamRequestInfo[ pixStreamId ][ localAssetId ] for this download
	local thisDownloadPixProviderRequests = downloadInfoForBinaryBucketId.pixProviderRequests

	for pixStreamId, localAssetIds in pairs( thisDownloadPixProviderRequests  ) do

		for localAssetId in pairs( localAssetIds ) do

			local activePixStreamRequestInfo =
					KSTableUtils_derefOpt( self._activePixStreamRequestInfo, 
											pixStreamId, localAssetId )

			if activePixStreamRequestInfo then

				self._activePixStreamRequestInfo[ pixStreamId ][ localAssetId ] = nil

			end

		end

	end

	-- 3) send the popped data

	local localAssetId = ozAssetIdInfoMap.localAssetId

	return localAssetId, downloadInfoForBinaryBucketId

end

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

local function getActiveDownloadCount( self )

	local activeDownloadCount = 0

	for _, ozAssetIdInfoMap in pairs( self._ozAssetIdInfoMap ) do

		for _ in pairs( ozAssetIdInfoMap.binaryBucketIdDownloadInfoMap ) do

			activeDownloadCount = activeDownloadCount + 1

		end

	end

	return activeDownloadCount

end

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

function PendingDownloadInfoCachePrototype:canDownload()

	return getActiveDownloadCount( self ) < self._maxDownloadCount

end

--------------------------------------------------------------------------------
-- This cache keeps track of the revId <-> digest maps for each in flight
-- download request identified by the ( ozAssetId, binaryType, binarySize )
--
-- When the download completes, we can look up the digest via the revId
--------------------------------------------------------------------------------

function PendingDownloadInfoCachePrototype:addNewPendingDownload( 
			ozAssetId, binaryType, binarySize, localAssetId,
			cancelDownloadFunc, httpPriority, pixelProviderPriority, maybeRenderSize, maybeRevId,
			maybeDigest )

	if not self:canDownload() then

		logAndRaiseError( "Can't start a new download. There are %s active and only %s max allowed",
							getActiveDownloadCount( self ), self._maxDownloadCount )

	end
	
	local ozAssetIdInfoMap = self._ozAssetIdInfoMap[ ozAssetId ] or
									{ localAssetId = localAssetId, 
										binaryBucketIdDownloadInfoMap = {} }

	self._ozAssetIdInfoMap[ ozAssetId ] = ozAssetIdInfoMap

	local binaryBucketId = makeBinaryBucketId( binaryType, binarySize )

	if ozAssetIdInfoMap.binaryBucketIdDownloadInfoMap[ binaryBucketId ] ~= nil then

		logAndRaiseError( "There is already a download of id '%s' for assetId %s",
								binaryBucketId, localAssetId )

	end

	local revIdDigestMap = {}

	local revId = maybeRevId or binaryBucketId

	local digest = maybeDigest or binaryBucketId

	revIdDigestMap[ revId ] = digest

	ozAssetIdInfoMap.binaryBucketIdDownloadInfoMap[ binaryBucketId ] = 
			{ cancelDownloadFunc = cancelDownloadFunc,
				httpPriority = httpPriority,
				pixelProviderPriority = pixelProviderPriority,
				renderSize = maybeRenderSize,
				revIdDigestMap = revIdDigestMap, 
				pixProviderRequests = {} }

end

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

function PendingDownloadInfoCachePrototype:joinPendingDownload( 
			ozAssetId, binaryType, binarySize, pixStreamId, localAssetId )

	local _, _, downloadInfoForBinaryBucketId =
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	if not downloadInfoForBinaryBucketId then

		logAndRaiseError( "Request %s:%s Can't join download of %s because there is no active download",
								pixStreamId, localAssetId, binaryType )
		
	end

	-- keep track of the pixstream requests on the download info, this is used
	-- to count the number of pixtream requests per download

	downloadInfoForBinaryBucketId.pixProviderRequests[ pixStreamId ] =
		downloadInfoForBinaryBucketId.pixProviderRequests[ pixStreamId ] or {}

	downloadInfoForBinaryBucketId.pixProviderRequests[ pixStreamId ][ localAssetId ] = true

	-- keep track of the pixProviderRequests by pixStream. This is used to map
	-- a pix stream request back to the actual download
	self._activePixStreamRequestInfo[ pixStreamId ] =
		self._activePixStreamRequestInfo[ pixStreamId ] or {}

	self._activePixStreamRequestInfo[ pixStreamId ][ localAssetId ] = 
		{ ozAssetId, binaryType, binarySize }
	
end

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

function PendingDownloadInfoCachePrototype:cancelAndRemoveDownload( 
											reason, ozAssetId, ozBinaryType,
											ozBinarySizeName )

	logger:infof( "Canceling download of %s( %s ) for %s because %s", 
					ozBinaryType, ozBinarySizeName, ozAssetId, reason )

	local localAssetId, downloadInfoForBinaryBucketId =
			self:removePendingDownloadInfo( ozAssetId, ozBinaryType,
												ozBinarySizeName )

	if not localAssetId then

		return {}

	end

	if type( downloadInfoForBinaryBucketId ) ~= "table" then

		logger:warnf( "No download info for localAssetId %s of %s( %s ) for %s", 
						localAssetId, ozBinaryType, 
						ozBinarySizeName, ozAssetId )

		return {}

	end

	if type( downloadInfoForBinaryBucketId.cancelDownloadFunc ) ~= "function" or
			not downloadInfoForBinaryBucketId.cancelDownloadFunc() then

		logger:warnf( "Failed canceling download of localAssetId %s %s( %s ) for %s", 
						localAssetId, ozBinaryType, ozBinarySizeName, 
						ozAssetId )

	end

	return type( downloadInfoForBinaryBucketId.pixProviderRequests ) == "table" and
			downloadInfoForBinaryBucketId.pixProviderRequests or {}

end

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

function PendingDownloadInfoCachePrototype:maybeGetActiveDownloadInfoForPixStreamRequest( 
											pixStreamId, localAssetId )

	local activePixStreamRequestInfo =
			KSTableUtils_derefOpt( self._activePixStreamRequestInfo, 
									pixStreamId, localAssetId )

	local ozAssetId, binaryType, binarySize = 
			unpack( activePixStreamRequestInfo or {} )

	if not ozAssetId then

		return

	end

	local ozAssetIdInfoMap, binaryBucketId, downloadInfoForBinaryBucketId = 
			maybeGetEntry( self, ozAssetId, binaryType, binarySize ) 

	return ozAssetId, binaryType, binarySize, ozAssetIdInfoMap, binaryBucketId,
			downloadInfoForBinaryBucketId

end

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

function PendingDownloadInfoCachePrototype:removePixStreamRequest( 
											pixStreamId, localAssetId, state )

	local ozAssetId, binaryType, binarySize, _, _,
			downloadInfoForBinaryBucketId =
				self:maybeGetActiveDownloadInfoForPixStreamRequest( 
						pixStreamId, localAssetId )

	if not ozAssetId then

		logger:warnf( "No activePixStreamRequestInfo for %s:%s with state %s (can happen if we delete pixel request and download completes before we successfully cancel download)", 
							pixStreamId, localAssetId, state )

		local activePixStreamRequestInfo =
				KSTableUtils_derefOpt( self._activePixStreamRequestInfo, 
										pixStreamId, localAssetId )

		if type( activePixStreamRequestInfo ) == "table" then

			self._activePixStreamRequestInfo[ pixStreamId ][ localAssetId ] = nil

		end

	end

	local pixProviderRequestsForPixStreamId =
			KSTableUtils_derefOpt( downloadInfoForBinaryBucketId,
									"pixProviderRequests", pixStreamId )

	if not pixProviderRequestsForPixStreamId then

		logger:warnf( "No pixProviderRequestsForPixStreamId for %s:%s with state %s (can happen if we delete pixel request and download completes before we successfully cancel download)",
						pixStreamId, localAssetId, state )

		return

	end

	pixProviderRequestsForPixStreamId[ localAssetId ] = nil

	local otherLocalAssetId = next( pixProviderRequestsForPixStreamId )

	if otherLocalAssetId then
		
		logger:infof( "Removed pix stream request %s:%s for active download, but download is still active on same pixStream for assetId %s",
						pixStreamId, localAssetId, otherLocalAssetId  )

		return

	end

	downloadInfoForBinaryBucketId.pixProviderRequests[ pixStreamId ] = nil

	local otherPixStreamId = next( downloadInfoForBinaryBucketId.pixProviderRequests ) 

	if otherPixStreamId then
		
		logger:infof( "Removed pix stream request %s:%s for active download, but download is still active another pixStream %s",
						pixStreamId, localAssetId, otherPixStreamId )

		return

	end

	-- There are no active pix provider requests for this download, remove it and cancel it.
	
	logger:infof( "Canceling active download for pix stream request %s:%s because it is the last active download for %s:%s ( %s )",
					pixStreamId, localAssetId, binaryType, binarySize,
					ozAssetId ) 

	self:cancelAndRemoveDownload( 
			"removed only pix provider download request", ozAssetId, binaryType, 
			binarySize ) 

end

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

local PendingDownloadInfoCache =
		Rx.Class( "PendingDownloadInfoCache", 
					PendingDownloadInfoCachePrototype )

--------------------------------------------------------------------------------
-- 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 getDownloadRequestInfo( 
				scheduler, revId, ozAssetId, ozBinaryType, ozBinarySizeName,
				baseRenderInfoCache, activeRenditionInfoCache,
				pendingDownloadInfoCache )
						
	local localAssetId, downloadInfoForBinaryBucketId =
			pendingDownloadInfoCache:removePendingDownloadInfo( 
										ozAssetId, ozBinaryType,
										ozBinarySizeName )

	if not localAssetId then

		return

	end
	
	logger:infof( "Removed request info %s( %s ) for completed download of %s",
						ozBinaryType, ozBinarySizeName, ozAssetId  )
						
	local requestsToUpdate = {}

	local maybeDigest
	
	if ( ozBinaryType == 'original' or ozBinaryType == 'proxy' ) then
		
		requestsToUpdate = PixelProviderSharedBlobCache.getActiveNegativeDownloadRequests( 
						scheduler, localAssetId, ozBinaryType )
		
	elseif ozBinaryType == 'rendition' then

		requestsToUpdate, maybeDigest = 
			getAffectedRenditionDownloadRequests( 
				scheduler, localAssetId, revId, baseRenderInfoCache,
				activeRenditionInfoCache, 
				downloadInfoForBinaryBucketId.renderSize, 
				downloadInfoForBinaryBucketId.revIdDigestMap )
				
	elseif ozBinaryType == 'xmp_develop' then
	
		requestsToUpdate = PixelProviderSharedBlobCache.getActiveXmpDevelopDownloadRequests( 
						scheduler, localAssetId, ozBinaryType )
		
	elseif ozBinaryType == 'xmp_sidecar' then

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

	end
	
	return localAssetId, requestsToUpdate, maybeDigest

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
	local decodeOk, maybeJson = pcall( require 'KSJsonDecoder'.decodeWebSafeJson, errorMessage )
	if decodeOk then
		ozCode = type( maybeJson ) == 'table' and tonumber( maybeJson.code )
	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 = 
			getDownloadRequestInfo( 
				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 )
		
	logger:errorf( "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, maybeDigest = 
			getDownloadRequestInfo( 
				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
	
	local transition

	if binaryBlob then
		
		logger:infof( "Got download %s %s for ozAssetId %s ( localAssetId %s )",
						ozBinaryType, ozBinarySizeName, ozAssetId, localAssetId )

		pendingSaveBlobCache
			:pushBlob( 
				localAssetId, binaryBlob, ozBinaryType, ozBinarySizeName,
				maybeDigest )
		
		transition = 'received_download'
	
	else
		-- 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 )

		transition = 'retry_download'
	
	end
	
	return requestsToUpdate, transition
	
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, pendingSaveBlobCache,
							maxConcurrentDownloads )

	local pendingDownloadInfoCache = PendingDownloadInfoCache( maxConcurrentDownloads )

	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 getFulfillableRequestsAndDownloadInfoForNegative( 
				scheduler, localAssetId, baseRenderInfoCache, ozBinaryType )

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

	local ozBinarySubtype = getBinarySubtypeForBinaryType(
														localAssetId,
														ozBinaryType,
														baseRenderInfoCache )

	return 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 )

	if PixelProviderSharedBlobCache.isNeedsRenditionDownloadState(
											downloadState ) then

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

	end

	if PixelProviderSharedBlobCache.isProxyDownloadState( downloadState ) then

		return true,
				getFulfillableRequestsAndDownloadInfoForNegative( 
					scheduler, localAssetId, baseRenderInfoCache, 'proxy' )

	end
	
	if PixelProviderSharedBlobCache.isOriginalDownloadState( 
											downloadState ) then

		return true,
				getFulfillableRequestsAndDownloadInfoForNegative( 
					scheduler, localAssetId, baseRenderInfoCache, 'original' )

	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 getCurrPixProviderPriorityForActiveDownloadRequests( 
					pendingDownloadPixProviderPriorities, pixProviderRequest,
					pendingDownloadInfoCache )

	local pixStreamId, localAssetId, _, currPixelProviderPriority =
			unpack( pixProviderRequest )

	local ozAssetId, binaryType, binarySize, _, binaryBucketId =
				pendingDownloadInfoCache:maybeGetActiveDownloadInfoForPixStreamRequest( 
						pixStreamId, localAssetId )

	if not binaryBucketId then

		return pendingDownloadPixProviderPriorities
		
	end

	local downloadInfoKey = string.format( "%s:%s", ozAssetId, binaryBucketId )

	local activeDownloadRequestInfo = 
			pendingDownloadPixProviderPriorities[ downloadInfoKey ] or
			{ currPixelProviderPriority, ozAssetId, binaryType, binarySize }

	-- set the pixel provider priority accordingly

	activeDownloadRequestInfo[ 1 ] =
		PixelPathProviderRequestTypes_comparePriorities(
			activeDownloadRequestInfo[ 1 ], 
			currPixelProviderPriority ) >= 0 and
		activeDownloadRequestInfo[ 1 ] or currPixelProviderPriority

	pendingDownloadPixProviderPriorities[ downloadInfoKey ] = 
		activeDownloadRequestInfo 

	return pendingDownloadPixProviderPriorities

end

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

local function extractValues( t )

	local values = {}

	for _, v in pairs( t ) do

		values[ #values + 1 ] = v

	end

	return values

end

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

local function getLowestPixProviderPriorityDownloadInfo( 
					lowestPriorityPreemptibleDownloadInfo, 
					currDownloadInfo,
					preemptingRequestPixProviderPriority )

	if PixelPathProviderRequestTypes_comparePriorities(
			currDownloadInfo[ 1 ],
			preemptingRequestPixProviderPriority ) >= 0 then

			-- this request can't be preempted, it has higher or equal priority
			-- than the preempting request.
			
		return lowestPriorityPreemptibleDownloadInfo

	end

	if not lowestPriorityPreemptibleDownloadInfo then

		-- this is the first preemptible download info we've seen
		return currDownloadInfo

	end

	if PixelPathProviderRequestTypes_comparePriorities(
			currDownloadInfo[ 1 ], 
			lowestPriorityPreemptibleDownloadInfo[ 1 ] ) < 0 then

		-- this request is lower priority
		return currDownloadInfo

	end

	return lowestPriorityPreemptibleDownloadInfo

end

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

local function cancelCurrentDownload( reason, scheduler,
										pendingDownloadInfoCache, ozAssetId,
										ozBinaryType, ozBinarySizeName )

	local pixProviderRequestsToRestart =
			pendingDownloadInfoCache
				:cancelAndRemoveDownload( 
					reason, ozAssetId, ozBinaryType, ozBinarySizeName )

	for pixStreamId, localAssetIds in pairs( pixProviderRequestsToRestart ) do

		for localAssetId in pairs( localAssetIds ) do
		
			scheduler:_maybeSetNextState( pixStreamId, localAssetId, 'reschedule_download' )

		end

	end

end

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

local function maybePreemptDownload( scheduler,
										preemptingRequestPixProviderPriority,
										pendingDownloadInfoCache )

	local downloadingRequests =
			scheduler:_getRequestsWithStatesInPriorityOrder(
						unpack( PixelProviderSharedBlobCache.downloadingBlobStates ) )

	local downloadInfoWithCurrPixProviderPriorities =
			WFFunctionUtils_reduce( {},
				getCurrPixProviderPriorityForActiveDownloadRequests, 
					downloadingRequests,
						pendingDownloadInfoCache )

	local preemptibleDownloadInfo =
			WFFunctionUtils_reduce( nil, 
				getLowestPixProviderPriorityDownloadInfo,
					extractValues( downloadInfoWithCurrPixProviderPriorities ),
					preemptingRequestPixProviderPriority )

	if not preemptibleDownloadInfo then

		return false

	end

	local preemptedPriority, ozAssetId, ozBinaryType, ozBinarySizeName = 
			unpack( preemptibleDownloadInfo )

	local cancelReason = 
			string.format( "request at prioirty %s preempted by higher priority",
							preemptedPriority )

	cancelCurrentDownload( cancelReason, scheduler, pendingDownloadInfoCache,
							ozAssetId, ozBinaryType, ozBinarySizeName )

	return true

end

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

local function mapPixelProviderPriorityToHttpPriority( currPixelProviderPriority )

	local priorityName = PixelPathProviderRequestTypes_getName(
								currPixelProviderPriority )

	if priorityName == PixelPathProviderRequestTypes_names.highest then
	
		return WFRxHttpClient.requestPriorityHighest
	
	end
	
	if priorityName == PixelPathProviderRequestTypes_names.grid_visible then

		return WFRxHttpClient.requestPriorityHigh

	end

	if priorityName == PixelPathProviderRequestTypes_names.lowest then
		
		return WFRxHttpClient.requestPriorityLow

	end

	return WFRxHttpClient.requestPriorityMedium

end

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

local function isHighHTTPPriority( httpPriority )

	return httpPriority == WFRxHttpClient.requestPriorityHigh or
			httpPriority == WFRxHttpClient.requestPriorityHighest

end

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

local function canJoinPendingSave( pendingSaveBlobCache, localAssetId, binaryType, 
												ozBinarySizeName, digest )
	
	return pendingSaveBlobCache:peekBlob( localAssetId, binaryType, 
												ozBinarySizeName, digest )

end

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

local function canStartOrJoinDownload( scheduler, pendingDownloadInfoCache,
											currPixelProviderPriority,
											ozAssetId, ozBinaryType,
											ozBinarySize ) 

	local maybeCurrHttpPriority =
			pendingDownloadInfoCache:getHttpPriorityIfDownloading( 
												ozAssetId, ozBinaryType,
												ozBinarySize ) 

	if maybeCurrHttpPriority then

		-- we are already downloading, see if we need to boost priority and
		-- warn that we won't if so.

		local newHttpPriority = mapPixelProviderPriorityToHttpPriority( 
									currPixelProviderPriority )

		if isHighHTTPPriority( newHttpPriority ) and WFRxHttpClient.isHigherPriority( newHttpPriority, maybeCurrHttpPriority ) then

			logger:warnf( "We have a '%s' pixel provider priority request for an existing download of %s %s for ozAssetId %s that was started at http priority %s. We need the highest http priority in order to make sure this request now takes precedent over all other http requests, so we are going to restart the current http request.",
				PixelPathProviderRequestTypes_getName( currPixelProviderPriority ),
				ozBinaryType, ozBinarySize, ozAssetId, maybeCurrHttpPriority )

			cancelCurrentDownload( 
				"restart at higher priority",
				scheduler, pendingDownloadInfoCache, ozAssetId, ozBinaryType,
				ozBinarySize )

		elseif maybeCurrHttpPriority ~= newHttpPriority and
			( maybeCurrHttpPriority == WFRxHttpClient.requestPriorityLow or
				( maybeCurrHttpPriority == WFRxHttpClient.requestPriorityMedium and
					( newHttpPriority == WFRxHttpClient.requestPriorityHigh or 
						newHttpPriority == WFRxHttpClient.requestPriorityHighest ) ) ) then

			logger:warnf( "We have a '%s' pixel provider priority request for an existing download of %s %s for ozAssetId %s which was started at http priority %s. The new http priority should be %s, which is higher, but we don't support updating the inflight http requests.",
				PixelPathProviderRequestTypes_getName( currPixelProviderPriority ),
				ozBinaryType, ozBinarySize, ozAssetId, maybeCurrHttpPriority,
				newHttpPriority )

		end

		return true

	end


	return pendingDownloadInfoCache:canDownload() or
			maybePreemptDownload( scheduler, currPixelProviderPriority, 
											pendingDownloadInfoCache ) 

end

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

local function lookForBinarySubtypeInList( ozBinarySubtype, ozBinaryList ) 

	ozBinaryList = ozBinaryList or {}
	
	for _, binaryInfo in ipairs( ozBinaryList ) do
		if binaryInfo.size == ozBinarySubtype then
			return binaryInfo.href
		end
	end
	
end

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

local function getBinaryUrlAndRevIdFromRenderInfo( maybeBaseRenderInfo, ozBinaryType, ozBinarySubtype )
	
	if type( maybeBaseRenderInfo ) ~= 'table' then

		return nil

	end
	
	local href

	if ozBinaryType == 'rendition' and ozBinarySubtype == "fullsize" then

		href = maybeBaseRenderInfo.downloadFullRenditionURI

	elseif ozBinaryType == 'rendition' then

		href = lookForBinarySubtypeInList( ozBinarySubtype, maybeBaseRenderInfo.renditions )

		if not href then

			href = lookForBinarySubtypeInList( ozBinarySubtype, maybeBaseRenderInfo.invalid_renditions )

		end

	elseif ozBinaryType == 'proxy' then

		href = lookForBinarySubtypeInList( ozBinarySubtype, maybeBaseRenderInfo.proxies )

	elseif ozBinaryType == 'original' then

		href = maybeBaseRenderInfo.ozOriginalLink

	elseif ozBinaryType == 'xmp_develop' then

		href = maybeBaseRenderInfo.ozXmpDevelopLink

	elseif ozBinaryType == 'xmp_sidecar' then

		href = maybeBaseRenderInfo.ozXmpLink

	end
	
	if not href then

		logger:warnf( "Unexpected ozBinaryType %s and ozBinarySubtype %s, no url hint provided to binary downloader.", 
				ozBinaryType, ozBinarySubtype )
		
		return

	end
	
	return href, maybeBaseRenderInfo.revId

end

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

local function startNewDownload( binaryRequester, baseRenderInfoCache,
									pendingDownloadInfoCache, ozAssetId,
									ozBinaryType, ozBinarySubtype,
									localAssetId, currPixelProviderPriority,
									renderSize, revId, digest )

	logger:infof( "About to start download %s %s for ozAssetId %s (localAssetId %s priority %s) at size: %s",
					ozBinaryType, ozBinarySubtype, ozAssetId,
					localAssetId, 
					PixelPathProviderRequestTypes_getName( currPixelProviderPriority ),
					renderSize )

	local httpPriority = mapPixelProviderPriorityToHttpPriority( currPixelProviderPriority )
	
	local maybeBaseRenderInfo = baseRenderInfoCache:peekCacheItem( localAssetId )

	local maybeBinaryUrl, maybeRevIdFromBaseRenderInfo = 
			getBinaryUrlAndRevIdFromRenderInfo( 
				maybeBaseRenderInfo, ozBinaryType, ozBinarySubtype )

	if not maybeBinaryUrl then

		return false, 
				string.format( "missing expected url for download of type: %s subtype: %s",
									ozBinaryType, ozBinarySubtype )

	end
	
	local cancelDownloadFunc =
			binaryRequester
				:maybeRequestBinaryAtUrl( 
					ozAssetId, ozBinaryType, ozBinarySubtype, maybeBinaryUrl, 
					revId or maybeRevIdFromBaseRenderInfo, httpPriority )

	if type( cancelDownloadFunc ) ~= "function" then

		return false, "binary requester didn't start a request"

	end

	pendingDownloadInfoCache
		:addNewPendingDownload( 
			ozAssetId, ozBinaryType, ozBinarySubtype, localAssetId,
			cancelDownloadFunc, httpPriority, currPixelProviderPriority,
			renderSize, revId, digest )

	return true

end

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

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

	local pixStreamId, localAssetId = unpack( request )

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

end

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

local function isDownloadBlocked( scheduler, ozBinaryType, maybeDownloadPriority )

	local statusTable = scheduler:getDaemonDownloadStatus()
	
	local catalogStatus = statusTable.catalogStatus
	local daemonDownloadStatus = statusTable.syncStatus
	local originalVolumeStatus = statusTable.originalVolumeStatus
	local proxyVolumeStatus = statusTable.proxyVolumeStatus
	local mainVolumeStatus = statusTable.mainVolumeStatus

	local maybePriorityName = maybeDownloadPriority 
			and PixelPathProviderRequestTypes_getName( maybeDownloadPriority )

	local downloadOriginalToMainDrive = false
	
	if catalogStatus == 'does_not_match' then
		
		return 'daemon', catalogStatus
	
	end
	
	if daemonDownloadStatus == 'offline' then
	
		-- We're unable to download anything when offline.
	
		return 'daemon', daemonDownloadStatus

	end
		
	if daemonDownloadStatus == 'sync_paused' and
			( ozBinaryType == 'original' or ozBinaryType == 'proxy' ) then

		-- When sync is paused, don't download any negatives.

		return 'daemon', daemonDownloadStatus
		
	end
	
	if ozBinaryType == 'original' and 
			( originalVolumeStatus == 'out_of_disk_space' or 
				originalVolumeStatus == 'volume_offline' ) then

		-- Download the original to the main drive when original volume is out of space or offline
		-- if export(ensure_local_original_exists) or loupe(highest) requests it.

		if PixelPathProviderRequestTypes_names.ensure_local_original_exists == maybePriorityName or
			PixelPathProviderRequestTypes_names.highest == maybePriorityName then

			downloadOriginalToMainDrive = true
			
		else
	
			-- Unable to download original because of an offline volume, or space constraints.
			return 'volume', originalVolumeStatus
		
		end

	end
	
	if ozBinaryType == 'proxy' and 
			( proxyVolumeStatus == 'out_of_disk_space' or 
				proxyVolumeStatus == 'volume_offline' ) then
		
		-- Unable to download proxy because of an offline volume, or space constraints.
	
		return 'volume', proxyVolumeStatus
	
	end
	
	if ( ozBinaryType == 'rendition' or 
				ozBinaryType == 'xmp_develop' or 
				downloadOriginalToMainDrive ) and 
			mainVolumeStatus == 'out_of_disk_space' then

		-- Unable to download renditions/xmp because of space constraints.
	
		return 'volume', mainVolumeStatus

	end
	
	return false

end

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

local function maybeCancelExistingDownloads( scheduler, pendingDownloadInfoCache, cancelDownloadStatusReason )
	
	local currentDownloadRequests = 
			pendingDownloadInfoCache:allActiveDownloadRequests()

	cancelDownloadStatusReason =
		string.format( "download task status: %s", cancelDownloadStatusReason )

	for _, downloadRequest in ipairs( currentDownloadRequests ) do
		
		local pixStreamId, localAssetId, ozAssetId, ozBinaryType, 
				ozBinarySize = unpack( downloadRequest )
		
		local pixelProviderPriority = 
				pendingDownloadInfoCache
					:getPixelProviderPriorityIfDownloading( 
						ozAssetId, ozBinaryType, ozBinarySize )
		
		local maybeBlockSource, maybeReason = 
				isDownloadBlocked( scheduler, ozBinaryType, pixelProviderPriority )
		
		if maybeBlockSource then
			
			logger:warnf( "Canceling download for %s:%s at priority %s because %s reported: %s",
							pixStreamId, localAssetId, pixelProviderPriority,  maybeBlockSource, maybeReason )

			cancelCurrentDownload( 
				cancelDownloadStatusReason, scheduler,
				pendingDownloadInfoCache, ozAssetId, ozBinaryType, 
				ozBinarySize )
			
		end
			
	end
	
end

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

local function maybeCancelExistingDownloadsIfBadDaemonStatus( 
				scheduler, pendingDownloadInfoCache )

	local statusTable = scheduler:getDaemonDownloadStatus()

	for downloadStatusType, status in pairs( statusTable ) do
	
		if status ~= 'okay' then

			local cancelReason =  
					string.format( "download status type %s has status %s",
										downloadStatusType, status )
		
			maybeCancelExistingDownloads( 
				scheduler, pendingDownloadInfoCache, cancelReason )
			
		end
		
	end

end

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

local PixelPathProviderSchedulerPrivateMethods_getDeletedRequestsWithState =
	PixelPathProviderSchedulerPrivateMethods.getDeletedRequestsWithState

--------------------------------------------------------------------------------
		
local function getNextPendingDownloadRequest( scheduler )

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

		
	local pixStreamId = unpack( request )

	if not pixStreamId then

		-- there were no requests to process
		return

	end

	return request

end

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

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

	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 )

	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 loopingOverSameRequest( request, prevRequest )

	if not prevRequest then

		return false

	end

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

	local prevPixStreamId, prevLocalAssetId, _, prevCurrPixelProviderPriority,
				prevDownloadState =
			unpack( prevRequest )

	if prevPixStreamId ~= pixStreamId or localAssetId ~= prevLocalAssetId then

		return false

	end

	logger:errorf( "Looper detected, processing the same request multiple times: prev state was ( %s, %s, priority: %s, state: %s ), new state is ( %s, %s, priority: %s, state: %s )",
					prevPixStreamId, prevLocalAssetId,
					prevCurrPixelProviderPriority, prevDownloadState,
					pixStreamId, localAssetId, currPixelProviderPriority,
					downloadState )

	return true

end

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

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

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

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

	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 = 
			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 == 'daemon' 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 canStartOrJoinDownload( 
			scheduler, pendingDownloadInfoCache, 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 =
			startNewDownload( 
				binaryRequester, baseRenderInfoCache, pendingDownloadInfoCache,
				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 )
	
	maybeCancelExistingDownloadsIfBadDaemonStatus( 
		scheduler, pendingDownloadInfoCache )

	local prevRequest
	
	while true do

		local request = getNextPendingDownloadRequest( scheduler )

		if not request then

			-- there are no requests to process, bail
			return

		end

		if loopingOverSameRequest( request, prevRequest ) then

			-- we have a looper, bail
			return

		end

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

			-- we can't process more requests (probably due to backlog), bail
			return

		end

		prevRequest = request

	end

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,

	}

}
