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

PixelProviderDownloadToFileTask.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2024 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.
------------------------------------------------------------------------------]]

	-- Rx
local Rx = require 'Rx'

	-- WF.core
local KSTableUtils = require 'KSTableUtils'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceUtils = require 'WFPerformanceUtils'
local WFCrypto = require 'WFCrypto'
	
	-- WF.oz_client.managed_catalog.pixel_provider
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local PixelProviderDownloadToFileUtils = 
		require 'PixelProviderDownloadToFileUtils'

	-- 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 WFCrypto_asyncDigestFile = assert( WFCrypto.asyncDigestFile,
			"WFCrypto.asyncDigestFile does not exist" )

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

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

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

local MAX_BINARY_DOWNLOAD_FAILURES_BEFORE_GIVING_UP = 6

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

local function isPermanentError( errorMessage, headers )
	
	-- get and check headers here
	local httpCode = headers and tonumber( headers.status_code ) or nil

	if not httpCode then

		-- if we don't have a status code, we can't do anything with this error
		return false

	end
	
	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 < 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
		
	local shouldRecordError = false

	if ( 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
	
	return shouldRecordError
	
end

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

local function handleFailedDownloadResult( 
				docStore, permanentBinaryDownloadFailureCounts,
				errorMessage, ozAssetId, ozBinaryType, ozBinarySizeName,
				headers, linksKey, binaryUrl )

	if errorMessage == 'canceledByNewRequest' then

		return 'reschedule_download'

	end
	
	local sawPermanentError = isPermanentError( errorMessage, headers )

	if sawPermanentError then

		permanentBinaryDownloadFailureCounts[ binaryUrl ] = 
			( permanentBinaryDownloadFailureCounts[ binaryUrl ] or 0 ) + 1

	end


	-- 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 ( permanentBinaryDownloadFailureCounts[ binaryUrl ] or 0 ) > 
			MAX_BINARY_DOWNLOAD_FAILURES_BEFORE_GIVING_UP then
	
		docStore:applyOverlay( 
			{ 
				_id = ozAssetId,
				_localOnly = { 
					binaryDownloadErrors = {
						[ linksKey ] = {
							[ binaryUrl ] = true
						}
					}
				}
			} )
			:call( require 'WFRxRunUntilDone' )
	end

	return sawPermanentError and 
			'download_to_file_fatal_fail' or 'download_to_file_transient_fail'
	
end

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

local function addLocalPathForRequestToLocalBinaryPathCache( 
				localBinaryPathCache, request, localPath, ozBinaryType )

	local pixStreamId, localAssetId = unpack( request )

	localBinaryPathCache
		:pushCacheItem( 
			pixStreamId, localAssetId, 
			{ localPath, ozBinaryType } )

	return localBinaryPathCache

end

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

local function makeBinaryRequestStreamHandler( 
				scheduler, docStore, pendingDownloadInfoCache, 
				localBinaryPathCache )

	local permanentBinaryDownloadFailureCounts = {}

	return function( okay, maybeLocalPathOrErrStr, ozAssetId,
						ozBinaryType, ozBinarySizeName, maybeFailureDetails )

		local localAssetId =
				pendingDownloadInfoCache
					:removePendingDownloadInfo( ozAssetId, ozBinaryType,
												ozBinarySizeName )

		if not localAssetId then

			scheduler:_sendTestData( "downloadCallbackWithoutLocalAssetId" )

			logger:warnf( "There is no pending download info for oz asset %s",
							ozAssetId )

			return

		end

		-- transition any pending needs download requests for this ozBinaryType
		-- and localAssetId
		PixelProviderDownloadToFileUtils.applyNeedsDownloadTransitionToPendingNeedDownloadLocalIdRequests(
			scheduler, ozBinaryType, localAssetId )

		-- transition any pending isDownloadingToFile requests for this ozBinaryType
		PixelProviderDownloadToFileUtils.applyStartTransitionToNeedsDownloadLocalIdRequests( 
			scheduler, ozBinaryType, localAssetId )
		
		local requestsToUpdate = 
				PixelProviderDownloadToFileUtils.getIsDownloadingToFileRequests( 
					scheduler, localAssetId, ozBinaryType )

		if type( requestsToUpdate ) ~= "table" then

			logger:errorf( "getIsDownloadingToFileRequests not supported for binary type: %s", ozBinaryType )

			return

		end

		if not next( requestsToUpdate ) then

			scheduler:_sendTestData( "downloadCallbackWithoutRequests" )

			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 )

			return

		end
			
		local failureHeaders = 
			KSTableUtils.derefOpt( maybeFailureDetails, 'headers' ) or {}

		local failureLinkKey = 
			KSTableUtils.derefOpt( maybeFailureDetails, 'linkKey' )

		local failureUrl = 
			KSTableUtils.derefOpt( maybeFailureDetails, 'url' )

		local transition =
			okay and
				'download_to_file_finished' or
				handleFailedDownloadResult( 
					docStore, permanentBinaryDownloadFailureCounts,
					maybeLocalPathOrErrStr, ozAssetId, ozBinaryType, ozBinarySizeName,
					failureHeaders, failureLinkKey, failureUrl )
									
		if not transition then

			return

		end

		if okay then

			-- each request gets a path entry in localBinaryPathCache
			local localPath =  maybeLocalPathOrErrStr

			localBinaryPathCache =
				WFFunctionUtils.reduce( 
					localBinaryPathCache,
					addLocalPathForRequestToLocalBinaryPathCache,
						requestsToUpdate, localPath, ozBinaryType )

		end
			
		scheduler:_applyTransitionAllRequests( requestsToUpdate, transition )

		WFFunctionUtils_map( 
			function ( request )

				local pixStreamId, localId = unpack( request )

				-- send the applied transitions to the unit test
				scheduler
					:_sendTestData( transition, pixStreamId, localId )

			end,
			requestsToUpdate )

		scheduler:_sendTestData( "finished handling download result" )

	end

end

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

local function updateDocStoreWithLocalPathStream( docStore, localPath,
													ozAssetId, ozBinaryType,
													ozBinarySizeName,
													maybeOrigSHA256 )

	return PixelProviderSharedTaskUtils.updateAssetDocValueLocalOnlyNegativePathResultStream(
				docStore, ozAssetId, ozBinaryType, ozBinarySizeName, localPath,
				maybeOrigSHA256 )
			:catch( 
				function( errStr )

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

					return Rx.Observable.empty()

				end )
			:takeLast( 0 )

end

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

local function maybeSha256Stream( filePath )

	return WFCrypto_asyncDigestFile( filePath, 'SHA256' )
			:take( 1 )
			:catch( 
				function( errStr )

					-- this is extremely unlikely to occur but there is no harm
					-- in returning nil. 
					logger:warnf( "error in asyncDigestFile: %s", errStr )
		
					return Rx.Observable.just( nil )

				end ) 

end

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

local function setupBinaryRequesterSubscription( 
				docStore, binaryRequester, destPathErrorStream,
				binaryRequestStreamHandler )

	local downloadToFileResultStream =
		Rx.Observable.merge( 
			binaryRequester:getBinaryRequestToFileStream(),
			destPathErrorStream )

	return downloadToFileResultStream
			:flatMap( 
				function( okay, filePathOrErrStr, ozAssetId, ozBinaryType,
									ozBinarySizeName, maybeFailureDetails )

					local resultStream =
							Rx.Observable.just( 
								okay, filePathOrErrStr, ozAssetId,
								ozBinaryType, ozBinarySizeName,
								maybeFailureDetails )

					if not okay then

						-- return the raw results immediately
						return resultStream

					end

					-- otherwise, update the doc store with the local path info
					-- before returning raw results
					local filePath = filePathOrErrStr 

					local maybeOrigSHA256Stream =
							ozBinaryType == 'original' and
								maybeSha256Stream( filePath ) or
								Rx.Observable.just( nil )

					local updateDocStoreWithLocalPathCompletionStream =
							maybeOrigSHA256Stream
								:flatMap( 
									function( maybeOrigSHA256 )

										return updateDocStoreWithLocalPathStream( 
												docStore, filePathOrErrStr,
												ozAssetId, ozBinaryType,
												ozBinarySizeName,
												maybeOrigSHA256 )

									end ) 
								:takeLast( 0 )

					return Rx.Observable.concat( 
							updateDocStoreWithLocalPathCompletionStream,
							resultStream )

				end )
			:subscribe( 
				function( okay, filePathOrErrStr, ozAssetId, ozBinaryType,
							ozBinarySizeName, maybeFailureDetails )

					local ok, maybeError =
							pcall( binaryRequestStreamHandler, 
									okay, filePathOrErrStr, ozAssetId,
									ozBinaryType, ozBinarySizeName,
									maybeFailureDetails )

					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, docStore, fileSpecObj,
							pendingDownloadInfoCache, localBinaryPathCache )

	local binaryRequester = 
		WFBinaryRequester( WFCatBinaryDownloader( wfCatalog ) )

	local taskState = {

		docStore = docStore,

		fileSpecObj = fileSpecObj,
		
		pendingDownloadInfoCache = pendingDownloadInfoCache,

		binaryRequester = binaryRequester,

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

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

	taskState.binaryRequesterSubscription =
		setupBinaryRequesterSubscription( 
			docStore, 
			binaryRequester, 
			pendingDownloadInfoCache:downloadToFileDestPathErrorStream(),
			taskState.binaryRequestStreamHandler )

	return taskState

end

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

local function removePixStreamRequestFromPendingDownloadCache( 
				deletedPixProviderRequest, pendingDownloadInfoCache )

	local pixStreamId, localAssetId, state = 
			unpack( deletedPixProviderRequest )

	pendingDownloadInfoCache
		:removePixStreamRequest( pixStreamId, localAssetId, state )

end

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

local function cleanupDeletedDownloadToFileRequests(
				scheduler, pendingDownloadInfoCache )

	local deletedIsDownloadingRequests =
			PixelPathProviderSchedulerPrivateMethods.getDeletedRequestsWithState(
				scheduler,
				unpack( PixelProviderDownloadToFileUtils.getIsDownloadingToFileStates( 
							scheduler ) ) )
	
	WFFunctionUtils_map( 
		removePixStreamRequestFromPendingDownloadCache, 
			deletedIsDownloadingRequests,
			pendingDownloadInfoCache )

end

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

local function addRequestToPendingDownloadInfoCache( 
				needsDownloadToFileRequest, pendingDownloadInfoCache,
				ozAssetId, ozBinaryType, ozBinarySubtype )

	local pixStreamId, localAssetId = unpack( needsDownloadToFileRequest )

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

end

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

local function maybeStartNewDownload(
				docStore, fileSpecObj, binaryRequester,
				baseRenderInfoCache, pendingDownloadInfoCache, localAssetId,
				ozAssetId, ozBinaryType, ozBinarySubtype,
				currPixelProviderPriority )

	local canonicalLocalPathStream =
		PixelProviderSharedTaskUtils.canonicalAssetNegativeLocalBinaryPathStream( 
			docStore, fileSpecObj, localAssetId, ozBinaryType, 
			ozBinarySubtype )

	local function downloadToFileViaBinaryRequester( 
					destPath, anOzAssetId, anOzBinaryType, anOzBinarySubtype,
					binaryUrl, httpPriority )

		return binaryRequester
				:maybeRequestBinaryAtUrlToFile( 
					destPath, anOzAssetId, anOzBinaryType, anOzBinarySubtype,
					binaryUrl, httpPriority )

	end

	local haveActiveDownloadRequest, failedToDownloadReason =
		pendingDownloadInfoCache
			:startNewDownloadToFile( 
				downloadToFileViaBinaryRequester, canonicalLocalPathStream,
				baseRenderInfoCache, localAssetId, ozAssetId, ozBinaryType,
				ozBinarySubtype, currPixelProviderPriority )

	return haveActiveDownloadRequest, failedToDownloadReason

end

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

local function applyMissingDownloadInfoTransition( 
				scheduler, needsDownloadToFileRequest, pixStreamId, localAssetId,
				errStr )

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

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

	scheduler
		:_applyTransitionToRequest( 
			needsDownloadToFileRequest, 'missing_download_info' )

end

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

local function joinIsDownloadingRequests( scheduler, pendingDownloadInfoCache )

	local isDownloadingStates =
			PixelProviderDownloadToFileUtils.getIsDownloadingToFileStates( 
				scheduler )

	-- look at all active requests pixel provider is downloading to file
	-- requests
	local activeDownloadOzBinaryTypeLocalIdMaps =
		PixelPathProviderSchedulerPrivateMethods.reduceAllRequestsWithStates( 
			scheduler, {}, isDownloadingStates, 
			function( ozBinaryTypeToLocalId, isDownloadingToFileRequest )

				local pixStreamId, localAssetId = 
					unpack( isDownloadingToFileRequest )

				-- is there actually an WFBinaryRequester request for this
				-- request?
				local ozAssetId, ozBinaryType, ozBinarySubtype =
					pendingDownloadInfoCache
						:maybeGetActiveDownloadInfoForPixStreamRequest( 
							pixStreamId, localAssetId )

				if not ozAssetId then

					return ozBinaryTypeToLocalId

				end

				local httpPriority =
					pendingDownloadInfoCache
						:getHttpPriorityIfDownloading( 
							ozAssetId, ozBinaryType, ozBinarySubtype )

				if not httpPriority then

					return ozBinaryTypeToLocalId

				end

				ozBinaryTypeToLocalId[ ozBinaryType ] = 
					ozBinaryTypeToLocalId[ ozBinaryType ] or {}

				ozBinaryTypeToLocalId[ ozBinaryType ][ localAssetId ] = 
					{ ozBinarySubtype, ozAssetId, httpPriority }

				return ozBinaryTypeToLocalId

			end )

	-- given the active downloads, find all requests that need a download for the 
	-- same asset and binary type at the same or lower http priority

	for ozBinaryType, localIdMap in pairs( activeDownloadOzBinaryTypeLocalIdMaps ) do

		for localAssetId, subtypeOzAssetIdTuple in pairs( localIdMap ) do

			local ozBinarySubtype, ozAssetId, httpPriority = unpack( subtypeOzAssetIdTuple )

			-- make sure we transition any pending needs download state requests first
			PixelProviderDownloadToFileUtils.applyNeedsDownloadTransitionToPendingNeedDownloadLocalIdRequests( 
				scheduler, ozBinaryType, localAssetId )

			-- now gather the needs download requests for this asset and binary type
			local currNeedsDownloadToFileRequestsForLocalAssetId =
					PixelProviderDownloadToFileUtils.maybeGetNeedsDownloadToFileRequests(
						scheduler, ozBinaryType, localAssetId ) or {}

			-- filter these requests to only those that will NOT preempt the active
			-- request
			local currNeedsDownloadToFileRequestsForLocalAssetIdToUpdate =
				WFFunctionUtils.filter( 
					function( request )

						local _, _, priority = unpack( request )

						local requestShouldPreemptExistingDownload =
								pendingDownloadInfoCache
									:pixelProviderPriorityShouldPreemptHttpPriority( 
										priority, httpPriority )

						if not requestShouldPreemptExistingDownload then

							return request

						end

					end, 
					currNeedsDownloadToFileRequestsForLocalAssetId )

			-- transition all these requests to is downloading
			PixelProviderDownloadToFileUtils.applyStartTransitionToNeedsDownloadRequests(
				scheduler, ozBinaryType, 
				currNeedsDownloadToFileRequestsForLocalAssetIdToUpdate )

			-- join the requests to the active download in the pending download
			-- info cache
			WFFunctionUtils_map( 
				addRequestToPendingDownloadInfoCache,
					currNeedsDownloadToFileRequestsForLocalAssetIdToUpdate,
					pendingDownloadInfoCache, ozAssetId,
					ozBinaryType, ozBinarySubtype )

		end

	end

end

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

local function processRequests( scheduler, taskState, baseRenderInfoCache )

	local pendingDownloadInfoCache = taskState.pendingDownloadInfoCache

	cleanupDeletedDownloadToFileRequests( scheduler, pendingDownloadInfoCache )
	
	pendingDownloadInfoCache
		:maybeCancelExistingDownloadsIfBadStatus( scheduler )

	joinIsDownloadingRequests( scheduler, pendingDownloadInfoCache )

	local needsDownloadToFileRequest = 
			pendingDownloadInfoCache
				:getNextDownloadToFileRequest( scheduler )

	if not needsDownloadToFileRequest then

		-- there are no requests to process, bail
		return

	end

	local pixStreamId, localAssetId, preferredSize, currPixelProviderPriority,
				downloadToFileState = unpack( needsDownloadToFileRequest )

	local ozAssetId = 
			KSTableUtils.derefOpt( 
				baseRenderInfoCache:peekCacheItem( localAssetId ), 
				'ozAssetId' )

	if not ozAssetId then

		return applyMissingDownloadInfoTransition( 
				scheduler, needsDownloadToFileRequest, pixStreamId,
				localAssetId, "no ozAssetId" )

	end

	------------------------------------------------------------
	-- 1) do we have all the info we need to download to file?
	------------------------------------------------------------
	local ozBinaryType = 
		PixelProviderDownloadToFileUtils.maybeGetBinaryTypeForDownloadToFileState( 
			scheduler, downloadToFileState )

	if not ozBinaryType then

		return applyMissingDownloadInfoTransition( 
				scheduler, needsDownloadToFileRequest, pixStreamId,
				localAssetId, "unable to determine binary type" )

	end

	local ozBinarySubtype =
			( ozBinaryType == 'original' and 'master' ) or
			( ozBinaryType == 'proxy' and
				PixelProviderSharedTaskUtils.maybeGetBinarySubtypeForProxy( 
					baseRenderInfoCache, localAssetId, preferredSize ) ) or
			nil

	if not ozBinarySubtype then

		local errStr = string.format( "unable to oz binary subtype for %s with size",
										ozBinaryType, preferredSize )

		return applyMissingDownloadInfoTransition( 
				scheduler, needsDownloadToFileRequest, pixStreamId,
				localAssetId, errStr )

	end


	----------------------------------------------------------------------------
	-- 2) get all needsDownload requests for localAssetId and binary type. 
	----------------------------------------------------------------------------

	-- Be sure to transition any pending needs download state requests before
	-- we collect all current needs download requests.
	PixelProviderDownloadToFileUtils.applyNeedsDownloadTransitionToPendingNeedDownloadLocalIdRequests( 
		scheduler, ozBinaryType, localAssetId )

	-- get all needsDownload requests for localAssetId and binary type, we
	-- expect this to contain the needsDownloadToFileRequest
	local needsDownloadToFileRequests =
			PixelProviderDownloadToFileUtils.maybeGetNeedsDownloadToFileRequests(
				scheduler, ozBinaryType, localAssetId )

	if type( needsDownloadToFileRequests ) ~= "table" then

		logger:errorf( "Skipping download for %s:%s because bad return value from maybeGetNeedsDownloadToFileRequests",
						pixStreamId, localAssetId )

		scheduler
			:_applyTransitionToRequest( 
				needsDownloadToFileRequest, 'cannot_download' )

		return

	end

	------------------------------------------------------------
	-- 3) Is downloading blocked for this (or all) needsDownloadToFileRequest?
	------------------------------------------------------------
	local downloadBlockedCategory =
			pendingDownloadInfoCache:isDownloadBlocked( 
				scheduler, ozBinaryType, currPixelProviderPriority )

	if downloadBlockedCategory then

		local downloadFailureTransition = 
				downloadBlockedCategory == 'wf' and 
					'no_internet' or 'volume_cannot_save'

		scheduler
			:_applyTransitionToRequest( 
				needsDownloadToFileRequest, downloadFailureTransition )

		return

	end

	------------------------------------------------------------
	-- 4) Can we start/join/preempt current dwnld for needsDownloadToFileRequest?
	------------------------------------------------------------
	local willProbablyPreemptDownload = 
			not pendingDownloadInfoCache:canDownload()

	if willProbablyPreemptDownload then

		-- send message to unit test
		scheduler
			:_sendTestData( 'reschedulingDownload' )

	end

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

		-- send message to unit test that we won't download 
		scheduler
			:_sendTestData( 'skipDownloadDueToBacklog' )

		return

	end
	
	------------------------------------------------------------
	-- 5) start/join/preempt current dwnld for needsDownloadToFileRequest?
	------------------------------------------------------------
	local haveActiveDownloadRequest =
			pendingDownloadInfoCache
				:hasEntry( ozAssetId, ozBinaryType, ozBinarySubtype )
										
	local failedToDownloadReason

	if not haveActiveDownloadRequest then

		-- start a new download for needsDownloadToFileRequest
		haveActiveDownloadRequest, failedToDownloadReason =
			maybeStartNewDownload(
				taskState.docStore, taskState.fileSpecObj, 
				taskState.binaryRequester,
				baseRenderInfoCache, pendingDownloadInfoCache, localAssetId,
				ozAssetId, ozBinaryType, ozBinarySubtype,
				currPixelProviderPriority )

	end

	if not haveActiveDownloadRequest then

		-- this is unexpected: we should have either joined or started a
		-- download at this point

		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 ),
						failedToDownloadReason ) 

		scheduler
			:_applyTransitionToRequest( 
				needsDownloadToFileRequest, 'cannot_download' )

		return

	end

	-- link all pixel provider needsDownloadToFileRequests to the actual
	-- WFBinaryRequester request. There is an n:1 relationship between pixel
	-- provider requests and WFBinaryRequester requests
	WFFunctionUtils_map( 
		addRequestToPendingDownloadInfoCache,
			needsDownloadToFileRequests, 
			pendingDownloadInfoCache, ozAssetId,
			ozBinaryType, ozBinarySubtype )

	-- transition all needsDownloadToFileRequests
	local maybeStartDownloadTransition = 
			PixelProviderDownloadToFileUtils.getStartDownloadingTransition( 
					ozBinaryType ) or
			'cannot_download'

	if maybeStartDownloadTransition == 'cannot_download' then

		logger:errorf( "no start download transition for %s", ozBinaryType )

	end

	scheduler
		:_applyTransitionAllRequests( 
			needsDownloadToFileRequests, 
			maybeStartDownloadTransition )

end

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

local function cleanup( taskState )
	
	if taskState.binaryRequesterSubscription then

		taskState.binaryRequesterSubscription:dispose()
		
		taskState.binaryRequesterSubscription = nil

	end

	if taskState.binaryRequester then

		taskState.binaryRequester:close()

		taskState.binaryRequester = nil

	end
	
end

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

local function injectMockBinaryRequester( taskState, maxDownloads,
												mockBinaryRequester )

	cleanup( taskState )

	taskState.binaryRequester = mockBinaryRequester

	taskState.binaryRequesterSubscription =
		setupBinaryRequesterSubscription( 
			taskState.docStore, 
			mockBinaryRequester,
			taskState.pendingDownloadInfoCache:downloadToFileDestPathErrorStream(),
			taskState.binaryRequestStreamHandler )

	taskState.pendingDownloadInfoCache:setMaxDownloadCount( maxDownloads )

	taskState.pendingDownloadInfoCache:reset()

	return taskState

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup,

	privateUnitTestExport = {

		injectMockBinaryRequester = injectMockBinaryRequester,

	}

}
