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

PixelProviderSaveDownloadedTask.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'

	-- pixel provider
local PixelProviderSharedBlobCache = 
		require 'PixelProviderSharedBlobCache'
local WFCatPixelFileCacheSave = require 'WFCatPixelFileCacheSave'
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"

	-- WF.core.utils
local WFAnalytics = require 'WFAnalytics'
local WFCrypto = require 'WFCrypto'
local WFFileUtils = require 'WFFileUtils'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceUtils = require 'WFPerformanceUtils'
local KSTableUtils = require 'KSTableUtils'
local KSData = require 'KSData'
local KSIdGenerator = require 'KSIdGenerator'
local KSPathUtils = require 'KSPathUtils'

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

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

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

local WFFunctionUtils_filter = assert( WFFunctionUtils.filter,
			"WFFunctionUtils.filter does not exist" )

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

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

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

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

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

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

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

local derefOpt = KSTableUtils.derefOpt

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

local function addLocalBinaryPathCacheItem( request, localBinaryPathCache,
												filePath, ozBinaryType )

	local pixStreamId, localAssetId = unpack( request )

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

end

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

local function sendTestMessage( request, scheduler, msg )

	local pixStreamId, localAssetId = unpack( request )

	scheduler:_sendTestData( msg, pixStreamId, localAssetId )

end

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

local _retryXmpDevelopSha256MismatchException = {}

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

local _fatalXmpDevelopSha256MismatchException = {}

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

local function getActiveRequestsThatRespondToTransition( 
					scheduler, localAssetId, transition )

	local srcStates = scheduler:_getSourceStatesForTransition( transition )

	if #srcStates == 0 then

		return {}

	end

	return scheduler:_getLocalAssetIdRequestsWithState( 
							localAssetId, unpack( srcStates ) ) 

end

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

local _xmpDevelopSha256MismatchReported = {}

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

local function sendXmpDevelopSha256MismatchAnalyticsError( docId, localAssetId )

	if _xmpDevelopSha256MismatchReported[ localAssetId ] then

		return

	end

	_xmpDevelopSha256MismatchReported[ localAssetId ] = true

	local errorEvent = {}

	errorEvent['event.type'] = 'error'

	errorEvent['event.appWorkflow'] = 'SYNC'

	errorEvent['event.subcategory'] = 'Sha256_mismatch'

	errorEvent['event.error_type'] = 'xmp_develop'

	errorEvent['event.error_desc'] = 'docValue sha256 does not match content sha256'

	errorEvent['content.id'] = docId or localAssetId

	errorEvent['content.subtype'] = 'image'

	WFAnalytics.trackEvent( errorEvent )

end

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

local function getErrorTransitionRequestsAndErrStr( scheduler, ozBinaryType, 
														localAssetId, errInfo )

	if errInfo == _retryXmpDevelopSha256MismatchException then

		local transition = "temp_xmp_develop_sha256_mismatch"

		local requests = getActiveRequestsThatRespondToTransition( 
							scheduler, localAssetId, transition )

		return transition, requests, "temp xmp_develop sha256 mismatch"

	end

	if errInfo == _fatalXmpDevelopSha256MismatchException then

		local transition = "fatal_xmp_develop_sha256_mismatch"

		local requests = getActiveRequestsThatRespondToTransition( 
							scheduler, localAssetId, transition )

		return transition, requests, "fatal xmp_develop sha256 mismatch"

	end

	local transition = 'save_downloaded_blob_failed'

	local errStr = errInfo

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

	return transition, requests, errStr

end

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

local function handleSaveBlobFailure( 
					scheduler, ozBinaryType, ozBinarySizeName, localAssetId,
					errInfo )

	local errTransition, requestsToUpdate, errStr =
			getErrorTransitionRequestsAndErrStr( 
				scheduler, ozBinaryType, localAssetId, errInfo )

	if #requestsToUpdate == 0 then

		logger:warnf( "We are ignoring a save failure of %s.%s for localAssetId %s because no requests are left that care about it.", 
							ozBinaryType, ozBinarySizeName, localAssetId )


		return

	end

	logger:warnf( "error saving %s.%s for localAssetId %s: %s",
					ozBinaryType, ozBinarySizeName, localAssetId, errStr )

	scheduler:_applyTransitionAllRequests( requestsToUpdate, errTransition )

	WFFunctionUtils_map( sendTestMessage, 
							requestsToUpdate, scheduler, 'saveFailed' )

end

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

local function handleSaveBlobSuccess( scheduler, localBinaryPathCache,
										ozBinaryType, ozBinarySizeName,
										localAssetId, filePath )

	local requestsToApplyBinarySaveTransition = 
			PixelProviderSharedBlobCache.getFulfilledBySuccessfulNegativeSaveRequests( 
				scheduler, localAssetId, ozBinaryType )

	if #requestsToApplyBinarySaveTransition == 0 then

		logger:warnf( "We are ignoring a save result of %s.%s for localAssetId %s because no requests are left that care about it.", 
							ozBinaryType, ozBinarySizeName, localAssetId )


		return

	end

	WFFunctionUtils_map( addLocalBinaryPathCacheItem, 
								requestsToApplyBinarySaveTransition, 
									localBinaryPathCache, 
									filePath, ozBinaryType )


	scheduler:_applyTransitionAllRequests( 
					requestsToApplyBinarySaveTransition,
					'saved_valid_download' )

	WFFunctionUtils_map( sendTestMessage, 
							requestsToApplyBinarySaveTransition, scheduler,
							'didPixelCacheSave' )
end

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

local function handleSaveBlobResult( scheduler, localBinaryPathCache, ok,
										filePathOrErrInfo, ozBinaryType,
										ozBinarySizeName, localAssetId )

	if not ok then

		local errInfo = filePathOrErrInfo

		handleSaveBlobFailure( 
			scheduler, ozBinaryType, ozBinarySizeName, localAssetId, errInfo )

		return

	end

	local localBinaryPath = filePathOrErrInfo

	handleSaveBlobSuccess( 
		scheduler, localBinaryPathCache, ozBinaryType, ozBinarySizeName,
		localAssetId, localBinaryPath )

end

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

handleSaveBlobResult = 
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription(
		"PixelProviderSaveDownloadedTask", 1 / 6, "handleSaveBlobResult", 
		handleSaveBlobResult )

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

local function setupTask( scheduler, docStore, fileSpecObj, 
							localBinaryPathCache, maxConcurrentSaves )

	local pendingSaveSubscriptions = {}

	local taskState = { maxConcurrentSaves = maxConcurrentSaves, 
						pendingSaveSubscriptions = pendingSaveSubscriptions,
						docStore = docStore,
						fileSpecObj = fileSpecObj }

	-- set up the save thread
	
	taskState.saveToDiskThread = WFCatPixelFileCacheSave( 
									"PixelProviderBinarySaver", 
									maxConcurrentSaves )

	-- set up the save result handler

	taskState.saveResultsStream = Rx.Subject()

	taskState.finishedSaveSubscription = 
	
		taskState.saveResultsStream:subscribe(
		
			function( ok, filePathOrErrInfo, pendingSaveNumber, ozBinaryType,
						ozBinarySizeName, binaryBlob, localAssetId )

				pendingSaveSubscriptions[ pendingSaveNumber ]:dispose()

				pendingSaveSubscriptions[ pendingSaveNumber ] = nil

				if binaryBlob then

					-- 20221020 bkruse
					--
					-- We used to retain the binaryBlob and pass it to handleSaveBlobResult in
					-- order to allow for a retry attempt on save. We no longer support
					-- retry logic, so release it here.
					--
					-- We should probably fix this to simply not retain the binaryBlob for
					-- the callback, but that has more risk.
					binaryBlob:close()

				end

				local ok2, maybeError =
						pcall( handleSaveBlobResult, 
								scheduler,
								localBinaryPathCache, 
								ok,
								filePathOrErrInfo, 
								ozBinaryType,
								ozBinarySizeName, 
								localAssetId )

				if not ok2 then

					logger:errorf( "error in handleSaveBlobResult: %s", maybeError )

				end

			end,
			
			function( errStr )

				logger:errorf( "pixel provider has permanently stopped handling save results due to taskState.saveResultsStream emitting error: %s", errStr )

			end )
			
	return taskState
			
end

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

local function maybeGetUnusedPendingSaveNumber( taskState )

	local pendingSaveNumber

	for i = 1, taskState.maxConcurrentSaves do

		pendingSaveNumber = 
			pendingSaveNumber or
			( not taskState.pendingSaveSubscriptions[ i ] and i ) or
			nil

	end

	return pendingSaveNumber

end

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

local function generateLocalFilePathForXmpDevelop( 
				fileSpecObj, xmpCameraRawSha256 )
	
	return xmpCameraRawSha256 and fileSpecObj:pathForSettingsXmp( xmpCameraRawSha256 )
	
end

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

local PixelProviderSharedTaskUtils_calculateLocalXmpSidecarPath =
		assert( PixelProviderSharedTaskUtils.calculateLocalXmpSidecarPath,
				"missing PixelProviderSharedTaskUtils.calculateLocalXmpSidecarPath" )

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

local function generateLocalFilePathForXmpSidecar( fileSpecObj, docValue )

	local assetId = derefOpt( docValue, '_id' )

	return assetId and 
			PixelProviderSharedTaskUtils_calculateLocalXmpSidecarPath(
				fileSpecObj, assetId )

end

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

local function logNonCanonicalPathError( ozBinaryType )

	if ozBinaryType == 'original' then
		-- By design originals must be uniquified by adding a numeric suffix.
		-- One example is duplicating doesn't change the importSource.fileName,
		-- but there may be other workflows that result in the same fileName.
		return
	end

	logger:warnf( "Using non canonical (uniquified) path for %s file",
						ozBinaryType )

end

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

local function makeFormattedErrStream( errStr, ... )

	return Rx.Observable.throw( string.format( errStr, ... ) )

end

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

local function writeNegativeToDiskStreamImp( saveToDiskThread, binaryBlob,
											filePath, localAssetId, 
											ozBinaryType )

	local fileSize = KSData.length( binaryBlob )
	
	if fileSize <= 0 then
	
	
		local errStr = string.format( "The %s data blob for localAssetId %s is empty",
											ozBinaryType, localAssetId )

		return Rx.Observable.just( false, errStr )
	
	end

	-- Perform the actual save off the pixel provider thread. We copy the blob
	-- when we pass it to savePixels because savePixels will delete the blob
	-- regardless of success or failure. In the event of failure, we need the blob
	-- for subsequent retries.
	
	local blobCopy = KSData.make( binaryBlob )

	if not blobCopy then

		local errStr = string.format( "Unable to copy %s blob to save for localAssetId %s",
											ozBinaryType, localAssetId )

		return Rx.Observable.just( false, errStr )

	end

	return Rx.Observable.create(

		function( observer )

			-- savePixels only fires onError or onCompleted events, need to
			-- set up a subscription to turn these into onNextEvents
			local saveToDiskResultsSubscription =
					saveToDiskThread
						:savePixels( filePath, blobCopy )
						:subscribe(

							function()
								
								logger:warnf( "Unexpected onNext event when saving %s blob to %s save for localAssetId %s",
													ozBinaryType, filePath, localAssetId )

							end,

							function( errStr )

								observer:onNext( false, errStr )
						
							end,
							
							function()

								observer:onNext( true, filePath, fileSize )

							end
							
						)

			return saveToDiskResultsSubscription
					
		end )
	
end

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

local function moveFileToCanonicalLocalPathResultStream( 
					tempPath, canonicalLocalPath, localAssetId, ozBinaryType )
	
	-- FIXME 20221019 bkruse
	--
	-- This stream does not emit errors but also has inconsistent results.
	--
	-- In success case, the return value is
	--
	-- 		actualDestPath
	--
	-- In the failure case, the return value
	--
	-- 		false, errStr
	--
	-- The downstream code does not handle the false, errStr return correctly
	return WFAsyncFileUtils.moveToUniquelyNamedFile( 
				tempPath, canonicalLocalPath )
			:map(
				function( actualDestPath )

					if actualDestPath ~= canonicalLocalPath then
						-- 20221103 bkruse 
						--
						-- intentionally logging both an error and warning.
						--
						-- The logger error is for analytics and we only want unique
						-- messages per ozBinaryType
						--
						-- The logger warn is for diagnostic log so we can
						-- identify which image ran into the issue.
						logNonCanonicalPathError( ozBinaryType )

						logger:warnf( "Alternate destination chosen when moving %s blob for %s from %s to %s. Actual dest file was %s",
										ozBinaryType, localAssetId, 
										tempPath, canonicalLocalPath,
										actualDestPath )

					end

					return actualDestPath

				end )
			:catch( 
				function( err )

					-- Give more context to the moveToUniquelyNamedFile error
					-- then rethrow
					local errStr = 
						string.format( "Failed to move %s blob for %s from %s to %s. Error was %s",
										ozBinaryType, localAssetId, tempPath,
										canonicalLocalPath, err )
			
					return Rx.Observable.just( false, errStr )
				
				end )
	
end

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

local function updateNegativeDocStoreInfoBeforeMoveResultStream( docStore, localAssetId,
													filePath, fileSize,
													ozBinaryType, ozBinarySizeName,
													sha256 )

	if ozBinaryType ~= 'original' and ozBinaryType ~= 'proxy' then

		return makeFormattedErrStream(
				"Wrong binary type sent to updateNegativeDocStoreInfoBeforeMoveResultStream: %s", 
				ozBinaryType )

	end
													
	local volumeGuid = filePath and WFFileUtils.getVolumeGuidForPath( filePath )
	local relativeFilePath = filePath and WFFileUtils.removeVolumeFromPath( filePath )

	if volumeGuid 
		and type( volumeGuid ) == 'string'
		and relativeFilePath
		and type( relativeFilePath ) == 'string' then
	
		local localOnly = {
			files = {
				[ ozBinaryType ] = {
					[ tostring( ozBinarySizeName ) ] = {
						relativePath = relativeFilePath,
						volumeGuid = volumeGuid,
						size = fileSize,
						sha256 = sha256
					}
				}
			},

			-- if we successfully downloaded the negative just remove all the binary download errors
			-- so we retry the various proxy/renditions and/or xmp files if needed. One example where this
			-- can happen is if we referred to a space account that had gone into an expired state and
			-- it is no longer expired.

			-- Note that the pixel provider doesn't utilize this error when downloading renditions as
			-- it uses WFCatFileSpec.getUrlForLinkKey which doesn't factor in the error state.

			-- The other APIs used by the idle downloader do check binaryDownloadErrors
			-- as they use the APIs like getOriginalLinkFromDocValue, getProxyLinkFromDocValue, and
			-- getXmpDevelopLinkAndSha256FromDocValue which return nil if the error is present.

			binaryDownloadErrors = { _delete = true },
		}

		local overlay = { 
			_id = localAssetId,
			_localOnly = localOnly,
		}
					
		return docStore
				:applyOverlay( overlay ) 
			:take( 1 )
			
	else
	
		return Rx.Observable.throw( "Unable to create doc store annotation for file " .. filePath )
	
	end
		
end

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

local function updateDocStoreXmpDevelopInfoStream( docStore, localAssetId, 
							filePath, fileSize,
							binaryBlob )
	
	return WFCrypto.asyncDigestData( binaryBlob, 'SHA256' )
			:flatMap(
				function( sha256 )

					return PixelProviderSharedTaskUtils.updateDocStoreXmpDevelopInfoStream(
							docStore, localAssetId, filePath, fileSize, 
							sha256 )

				end )

end

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

local function updateDocStoreXmpSidecarInfoStream( docStore, localAssetId,
							filePath, binaryBlob )

	local localOnly = {
		xmpSidecarExists = true,
		files = {
			original = {
				xmp = {
					path = filePath,
					fileSize = KSData.length( binaryBlob ),
				}
			}
		}
	}

	local overlay = {
		_id = localAssetId,
		_localOnly = localOnly,
	}

	return docStore:applyOverlay( overlay )
		:take( 1 )

end

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

local function makeIgnoreAllDataStream( srcStream )

	return srcStream:takeLast( 0 )

end

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

local function writeBinaryBlobToTempSavePathResultStream( 
				canonicalLocalPath, fileSpecObj, saveToDiskThread, binaryBlob,
				localAssetId, ozBinaryType )

	-- Because we first download the asset to a temporary
	-- location on disk, we need the temporary location to
	-- be on the same volume as the final location. This
	-- makes the final file move operation much faster.
	
	-- TODO [jandrew 2017-05-05]: Before we trigger a new download, we should
	-- check to see if the asset lives in the temporary directory. If so, we
	-- can simply move it to it's final destination instead of redownloading it.
	
	local canonicalLocalPathVolumeGuid = canonicalLocalPath and WFFileUtils.getVolumeGuidForPath( canonicalLocalPath )
	
	local tempSaveFolderForVolume = fileSpecObj:tempFolderPathForVolume( canonicalLocalPathVolumeGuid )
	
	local saveFilePath = KSPathUtils.child( tempSaveFolderForVolume, KSIdGenerator.uuidInLowercaseHex() )
	
	-- Write the negative to a temporary location on disk.
	-- This stream emits
	--
	-- 		didSaveBlobOk, savedPathOrError, fileSize 
	return writeNegativeToDiskStreamImp( 
				saveToDiskThread, binaryBlob, saveFilePath,
				localAssetId, ozBinaryType )

end

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

local function makeNonEmptyOzBinaryDocValueStream( 
					docStore, localAssetId, ozBinaryType )

	return docStore
			:docValue( localAssetId )
			:flatMapLatest( 

				function( docValue )

					if docValue._empty then

						return makeFormattedErrStream( 
								"Binary type %s for localAssetId %s has been deleted",	
									ozBinaryType, localAssetId )

					end

					return Rx.Observable.just( docValue )

				end )
			:take( 1 )

end

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

local _mostRecentSha256Mismatch = KSTableUtils.makeWithMode( "k" )

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

local function maybeGetMostRecentSha256Mismatch( docStore, localAssetId )

	if not _mostRecentSha256Mismatch[ docStore ] then

		return

	end

	return _mostRecentSha256Mismatch[ docStore ][ localAssetId ]

end

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

local function setMostRecentSha256Mismatch( docStore, localAssetId, sha256 )

	_mostRecentSha256Mismatch[ docStore ] =
		_mostRecentSha256Mismatch[ docStore ] or {}
	
	_mostRecentSha256Mismatch[ docStore ][ localAssetId ] = sha256

end

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

local function clearMostRecentSha256Mismatch( docStore, localAssetId )

	setMostRecentSha256Mismatch( docStore, localAssetId, nil )

end

--------------------------------------------------------------------------------
-- 20230822 bkruse
--
-- Per NMB-25842, we are seeing an occasional sha256 mismatch on Oz. i.e. the
-- sha256 of the Oz xmp develop blob does not match the sha256 in the Oz asset
-- metadata. LrD is the only client that is checking that match and after
-- meeting with the Oz and WF teams, we jointly decided not to enforce the
-- match.
--
-- We will continue to check the match and log an error if it does not match,
-- but will no longer treat mismatch as fatal.
local TREAT_PERSISTENT_XMP_DEVELOP_SHA256_MISMATCH_AS_FATAL -- = true

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

local function getXmpDevelopSavePath( docStore, fileSpecObj, localAssetId,
										docValue, xmpBlobSha256 )

	if not xmpBlobSha256 then

		-- we should always have a sha256 for the blob,
		-- it's fatal if we do not
		--
		-- 20230628 bkruse
		-- This is exceedingly unlikely, therefore no analytics sent
		return false, _fatalXmpDevelopSha256MismatchException

	end

	local docValueXmpDevelopSha256 =
			derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' ) 

	if not docValueXmpDevelopSha256 then

		-- we should always have a docValue sha256, too,
		-- it's fatal if we do not
		--
		-- 20231204 bkruse
		-- This is also exceedingly unlikely, therefore no analytics sent
		return false, _fatalXmpDevelopSha256MismatchException

	end

	-- we need to save at the path that is based on the docValue sha256, not
	-- the actual sha256 of the file contents
	local docValueBasedXmpDevelopSavePath = 
			generateLocalFilePathForXmpDevelop( 
				fileSpecObj, docValueXmpDevelopSha256 )

	if xmpBlobSha256 == docValueXmpDevelopSha256 then

		clearMostRecentSha256Mismatch( docStore, localAssetId )

		return true, docValueBasedXmpDevelopSavePath

	end

	local mostRecentMismatchSha256 =
			maybeGetMostRecentSha256Mismatch( docStore, localAssetId ) 

	if xmpBlobSha256 ~= mostRecentMismatchSha256 then

		-- if this is the first mismatch, log and remember it
		setMostRecentSha256Mismatch( docStore, localAssetId, xmpBlobSha256 )

		-- this might be a race condition, so don't treat it as fatal.
		return false, _retryXmpDevelopSha256MismatchException

	end

	-- 20231204 bkruse
	--
	-- at this point the sha256 mismatch has been seen multiple times in a row
	-- sha256 mismatch on Oz is happening in the wild. See NMB-25484
	local docId = derefOpt( docValue, '_id' )

	-- send analytics
	sendXmpDevelopSha256MismatchAnalyticsError( docId, localAssetId )

	--
	-- we should either:
	-- 	a) treat it as fatal error (FATAL)
	-- 	b) just store the file with the mismatched sha256 at the expected path
	-- 	   (FAKE IT)

	if TREAT_PERSISTENT_XMP_DEVELOP_SHA256_MISMATCH_AS_FATAL then

		-- FATAL
		-- two mismatches in a row with same xmpBlobSha256 is likely a looper and
		-- should be treated as a fatal error
		return false, _fatalXmpDevelopSha256MismatchException

	end

	-- FAKE IT intentionally going to store the wrong data at the right path
	return true, docValueBasedXmpDevelopSavePath

end

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

local function validatedCanonicalXmpDevelopLocalFilePathStream( 
					docStore, nonEmptyOzBinaryDocValueStream, fileSpecObj,
					localAssetId, xmpBlobSha256 )

	return nonEmptyOzBinaryDocValueStream
			:flatMapLatest( 

				function( docValue )

					local ok, xmpDevelopSavePathOrErrStr =
						getXmpDevelopSavePath( 
							docStore, fileSpecObj, localAssetId, docValue,
							xmpBlobSha256 )

					if not ok then

						local errStr = xmpDevelopSavePathOrErrStr

						return Rx.Observable.throw( errStr )

					end

					local xmpDevelopSavePath = xmpDevelopSavePathOrErrStr
					
					return Rx.Observable.just( xmpDevelopSavePath )

				end )

end

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

local function canonicalSidecarLocalFilePathStream( 
				nonEmptyOzBinaryDocValueStream, fileSpecObj )

	return nonEmptyOzBinaryDocValueStream
			:map( 

				function( docValue )

					return generateLocalFilePathForXmpSidecar( 
							fileSpecObj, docValue ) 

				end )

end

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

local function canonicalXmpBlobLocalFilePathStream( 
					docStore, fileSpecObj, localAssetId, ozBinaryType, 
					maybeXmpBlobSha256 )

	local canonicalXmpBlobLocalFilePathStreamImp

	local nonEmptyOzBinaryDocValueStream =
			makeNonEmptyOzBinaryDocValueStream( 
				docStore, localAssetId, ozBinaryType )

	if ozBinaryType == 'xmp_develop' then

		canonicalXmpBlobLocalFilePathStreamImp =
			validatedCanonicalXmpDevelopLocalFilePathStream( 
					docStore, nonEmptyOzBinaryDocValueStream, fileSpecObj,
					localAssetId, maybeXmpBlobSha256 )

	end

	if ozBinaryType == 'xmp_sidecar' then

		canonicalXmpBlobLocalFilePathStreamImp =
			canonicalSidecarLocalFilePathStream( 
				nonEmptyOzBinaryDocValueStream, fileSpecObj )

	end

	if not canonicalXmpBlobLocalFilePathStreamImp then

		return makeFormattedErrStream( 
				"Wrong binary type sent to canonicalXmpBlobLocalFilePathStream: %s", 
					ozBinaryType )

	end

	return canonicalXmpBlobLocalFilePathStreamImp
			:flatMapLatest( 

				function( filePath )

					if not filePath then
					
						return makeFormattedErrStream( 
								"Unable to create a %s file path for localAssetId %s",
									ozBinaryType, localAssetId )
					
					end
					
					return Rx.Observable.just( filePath )

				end )

end


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

local function saveXmpBlobAsLocalFileResultStream( 
				docStore, fileSpecObj, saveToDiskThread, localAssetId,
				ozBinaryType, binaryBlob )

				-- 1) calculate the local path
	local maybeXmpBlobSha256Stream =
			( ozBinaryType == 'xmp_develop' and
				WFCrypto.asyncDigestData( binaryBlob, 'SHA256' ) ) or

			( ozBinaryType == 'xmp_sidecar' and
				Rx.Observable.just( nil ) ) or

			makeFormattedErrStream( 
				"Wrong binary type sent to saveXmpBlobAsLocalFileResultStream: %s", 
					ozBinaryType )

	return maybeXmpBlobSha256Stream
				:flatMapLatest(
					function( maybeXmpBlobSha256 )

						return canonicalXmpBlobLocalFilePathStream( 
								docStore, fileSpecObj, localAssetId, ozBinaryType,
								maybeXmpBlobSha256 )

					end )
				-- 2) write out the blob to temp save path
				:flatMapLatest( 
					function( canonicalLocalPath )
				
						local writeNegativeToDiskResultStream =
								writeBinaryBlobToTempSavePathResultStream( 
									canonicalLocalPath, fileSpecObj,
									saveToDiskThread, binaryBlob, localAssetId,
									ozBinaryType )

						return Rx.Observable.combineLatest(
								Rx.Observable.just( canonicalLocalPath ),
								writeNegativeToDiskResultStream )
					
					end )
				-- 3) if ok, update docstore to contain the temp save path
				:flatMapLatest( 
					function( canonicalLocalPath, didSaveBlobOk, savedPathOrError, fileSize )

						if not didSaveBlobOk then

							local errStr = savedPathOrError

							return Rx.Observable.throw( errStr )

						end

						local savedPath = savedPathOrError

						local rawUpdateDocStoreBeforeMoveResultStream =

							( ozBinaryType == 'xmp_develop' and
								updateDocStoreXmpDevelopInfoStream( 
									docStore, localAssetId, savedPath,
									fileSize, binaryBlob ) ) or

							( ozBinaryType == 'xmp_sidecar' and
								updateDocStoreXmpSidecarInfoStream(
									docStore, localAssetId, savedPath,
									binaryBlob ) ) or

							Rx.Observable.throw(
								string.format( "saveSharedSha256LocalBinaryResultStream invoked with bad ozBinaryType %s", 
													ozBinaryType ) )
						
						-- make sure the update operation just completes
						-- without emitting data
						local updateDocStoreBeforeMoveResultStream =
								makeIgnoreAllDataStream( 
									rawUpdateDocStoreBeforeMoveResultStream )

						-- This will emit a single tuple:
						--
						-- 		savedPath, canonicalLocalPath
						return Rx.Observable.concat(
								updateDocStoreBeforeMoveResultStream,
								Rx.Observable.just( savedPath, canonicalLocalPath, fileSize ) )
							
					end )
				-- 4) Move the negative to the final location.
				--
				--    The reason this step comes after creating the doc store
				--    annotation is because it prevents us from leaving
				--    untracked files on disk in the event that the app exits
				--    between these two steps.  Before, we were seeing this
				--    happen often when quitting the app during first time
				--    sync.
				:flatMapLatest( 
					function( savedPath, canonicalLocalPath, fileSize )

						-- this will emit a single value async:
						--
						-- 	actualDestPath
						return Rx.Observable.combineLatest(
								Rx.Observable.just( fileSize ),
								moveFileToCanonicalLocalPathResultStream( 
									savedPath, canonicalLocalPath, localAssetId,
									ozBinaryType ) )

					end )
				-- 5) If move succeeds, update docstore to point at
				--    canonicalLocalPath
				--
				:flatMapLatest( 
					function( fileSize, finalSavePath )

						local rawUpdateDocStoreAfterMoveResultStream =
							( ozBinaryType == 'xmp_develop' and
								updateDocStoreXmpDevelopInfoStream( 
									docStore, localAssetId, finalSavePath,
									fileSize, binaryBlob ) ) or

							( ozBinaryType == 'xmp_sidecar' and
								updateDocStoreXmpSidecarInfoStream(
									docStore, localAssetId, finalSavePath,
									binaryBlob ) ) or

							Rx.Observable.throw(
								string.format( "saveSharedSha256LocalBinaryResultStream invoked with bad ozBinaryType %s", 
													ozBinaryType ) )
									
						local updateDocStoreAfterMoveResultStream =
								makeIgnoreAllDataStream(
									rawUpdateDocStoreAfterMoveResultStream )

						return Rx.Observable.concat(
								updateDocStoreAfterMoveResultStream,
								Rx.Observable.just( finalSavePath ) )

					end )

	
end

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

local function saveAssetNegativeLocalBinaryResultStream( 
				docStore, fileSpecObj, saveToDiskThread, localAssetId,
				ozBinaryType, ozBinarySizeName, binaryBlob, maybeBlobSha256 )

				-- 1) calculate the local path
	return PixelProviderSharedTaskUtils.canonicalAssetNegativeLocalBinaryPathStream( 
					docStore, fileSpecObj, localAssetId, ozBinaryType,
					ozBinarySizeName, maybeBlobSha256 )
				-- 2) write out the blob to temp save path
				:flatMapLatest( 
					function( canonicalLocalPath, docStoreSha256 )

						local writeNegativeToDiskResultStream =
								writeBinaryBlobToTempSavePathResultStream( 
									canonicalLocalPath, fileSpecObj,
									saveToDiskThread, binaryBlob, localAssetId,
									ozBinaryType )

						local sha256 = maybeBlobSha256 or docStoreSha256

						return Rx.Observable.combineLatest(
								Rx.Observable.just( canonicalLocalPath ),
								Rx.Observable.just( sha256 ),
								writeNegativeToDiskResultStream )
					
					end )
				-- 3) if ok, update docstore to contain the temp save path
				:flatMapLatest( 
					function( canonicalLocalPath, sha256, didSaveBlobOk, savedPathOrError, fileSize )

						if not didSaveBlobOk then

							local errStr = savedPathOrError

							return Rx.Observable.throw( errStr )

						end

						local savedPath = savedPathOrError
						
						-- update the doc store with the save path, this
						-- depends on ozBinaryType
						local rawUpdateDocStoreBeforeMoveResultStream =
							-- proxies or origs
							updateNegativeDocStoreInfoBeforeMoveResultStream( 
								docStore, localAssetId, canonicalLocalPath, fileSize,
								ozBinaryType, ozBinarySizeName, sha256 )
						
						-- make sure the update operation just completes
						-- without emitting data
						local updateDocStoreBeforeMoveResultStream =
								makeIgnoreAllDataStream( 
									rawUpdateDocStoreBeforeMoveResultStream )

						-- This will emit a single tuple:
						--
						-- 		savedPath, canonicalLocalPath
						return Rx.Observable.concat(
								updateDocStoreBeforeMoveResultStream,
								Rx.Observable.just( savedPath, canonicalLocalPath ) )
							
					end )
				-- 4) Move the negative to the final location.
				--
				--    The reason this step comes after creating the doc store
				--    annotation is because it prevents us from leaving
				--    untracked files on disk in the event that the app exits
				--    between these two steps.  Before, we were seeing this
				--    happen often when quitting the app during first time
				--    sync.
				:flatMapLatest( 
					function( savedPath, canonicalLocalPath )

						-- this will emit a single value async:
						--
						-- 	actualDestPath
						return moveFileToCanonicalLocalPathResultStream( 
								savedPath, canonicalLocalPath, localAssetId,
								ozBinaryType )

					end )
				-- 5) If move succeeds, update docstore to point at
				--    canonicalLocalPath
				--
				--    FIXME [jandrew 2016-11-30]: Because the UI doesn't listen
				--    on the PP send data task for when to display images in
				--    Loupe, but rather watches the doc value of the asset,
				--    this final step it what triggers the UI to try to load
				--    the image now that it's in the proper location on disk.
				--    This should really be fixed so that the UI listens for
				--    the response from the PP, and doesn't simply watch the
				--    doc value. 
				:flatMapLatest( 
					function( finalSavePath )

						local rawUpdateDocStoreAfterMoveResultStream =
								PixelProviderSharedTaskUtils.updateAssetDocValueLocalOnlyNegativePathResultStream(
									docStore, localAssetId, ozBinaryType,
									ozBinarySizeName, finalSavePath ) 

						local updateDocStoreAfterMoveResultStream =
								makeIgnoreAllDataStream(
									rawUpdateDocStoreAfterMoveResultStream )

						return Rx.Observable.concat(
								updateDocStoreAfterMoveResultStream,
								Rx.Observable.just( finalSavePath ) )

					end )

end

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

local function saveLocalBinaryResultStream( 
				docStore, fileSpecObj, saveToDiskThread, localAssetId,
				ozBinaryType, ozBinarySizeName, binaryBlob, maybeBlobSha256 )

	if ozBinaryType == 'original' or ozBinaryType == 'proxy' then

		return saveAssetNegativeLocalBinaryResultStream(
				docStore, fileSpecObj, saveToDiskThread, localAssetId,
				ozBinaryType, ozBinarySizeName, binaryBlob, maybeBlobSha256 )

	end

	if ozBinaryType == 'xmp_develop' or ozBinaryType == 'xmp_sidecar' then

		return saveXmpBlobAsLocalFileResultStream(
				docStore, fileSpecObj, saveToDiskThread, localAssetId,
				ozBinaryType, binaryBlob )
	
	end

	return makeFormattedErrStream( 
			"Binary type not supported by PixelProviderSaveDownloadedTask: %s",
				ozBinaryType )
	
end

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

local function hasSaveNegativeRequests( localAssetId, scheduler )

	local saveNegativeRequests =	
			scheduler:_getLocalAssetIdRequestsWithState( 
				localAssetId,
				unpack( PixelProviderSharedBlobCache.saveRenderBinaryStates ) )

	return next( saveNegativeRequests ) and localAssetId or nil

end

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

local function hasNoSaveNegativeRequests( localAssetId, scheduler )

	return not hasSaveNegativeRequests( localAssetId, scheduler ) and
			localAssetId or nil

end

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

local function destroyAllLocalAssetIdNegatives( localAssetId, 
													pendingSaveBlobCache )

	pendingSaveBlobCache:destroyAllLocalAssetIdNegatives( localAssetId )

end

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

local function purgeAbandonedBlobs( scheduler, pendingSaveBlobCache,
										localAssetIdsWithBlob )

	local localAssetIdsWithBlobAndNoRequest =
			WFFunctionUtils_filter(
				hasNoSaveNegativeRequests,
					localAssetIdsWithBlob, scheduler )

	WFFunctionUtils_map( destroyAllLocalAssetIdNegatives,
							localAssetIdsWithBlobAndNoRequest, 
							pendingSaveBlobCache )

end

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

local function addActiveNegativeSaveInfo( activeNegativeSaves, request )

	local _, localAssetId, _, _, activeSaveState = unpack( request )

	activeNegativeSaves[ localAssetId ] = 
		activeNegativeSaves[ localAssetId ] or {}

	local ozBinaryType = 
			( PixelProviderSharedBlobCache.isProxyDownloadState( 
											activeSaveState ) and 'proxy' ) or 
			( PixelProviderSharedBlobCache.isOriginalDownloadState( 
											activeSaveState ) and 'original' ) or 
			nil

	if ozBinaryType then

		activeNegativeSaves[ localAssetId ][ ozBinaryType ] = true

	end

	return activeNegativeSaves

end

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

local function gatherActiveNegativeSavesInfo( activeNegativeSaves, localAssetId, 
											scheduler )

	local requestsWithActiveNegativeSave =
			scheduler:_getLocalAssetIdRequestsWithState( 
				localAssetId, 
				unpack( PixelProviderSharedBlobCache.savingRenderBinaryStates ) )

	return WFFunctionUtils_reduce( activeNegativeSaves, 
									addActiveNegativeSaveInfo,
										requestsWithActiveNegativeSave )

end

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

local function joinRequestsToActiveSaveRequests( 
							scheduler, localAssetIdsWithBlob )
			
	local activeNegativeSavesInfo = WFFunctionUtils_reduce( {}, 
										gatherActiveNegativeSavesInfo, 
											localAssetIdsWithBlob, scheduler )

	for localAssetId, activeSaveOzBinaryTypes in pairs( activeNegativeSavesInfo ) do

		for activeSaveOzBinaryType in pairs( activeSaveOzBinaryTypes ) do

			local canJoinActiveSaveRequests = 
					PixelProviderSharedBlobCache.getCanJoinNegativeSaveRequests( 
						scheduler, localAssetId, activeSaveOzBinaryType )

			WFFunctionUtils_map( sendTestMessage, 
									canJoinActiveSaveRequests,
									scheduler, 'joinActiveSave' )

			scheduler:_applyTransitionAllRequests( 
							canJoinActiveSaveRequests,
							'have_pending_save_for_blob' )

		end

	end

end

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

local PixelProviderSharedTaskUtils_maybeGetBinarySubtypeForProxy =
		PixelProviderSharedTaskUtils.maybeGetBinarySubtypeForProxy

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

local function updateRequestAndMaybeSubscribeToSave( 
				taskState, scheduler, baseRenderInfoCache,
				pendingSaveBlobCache, pendingSaveNumber,
				needToSaveRenderBinaryRequest )

	local pixStreamId, localAssetId, preferredSize, _, readyToSaveState = unpack( needToSaveRenderBinaryRequest )

	local ozBinaryType = 
			( PixelProviderSharedBlobCache.isProxyDownloadState( 
											readyToSaveState ) and 'proxy' ) or 
			( PixelProviderSharedBlobCache.isOriginalDownloadState( 
											readyToSaveState ) and 'original' ) or 
			( PixelProviderSharedBlobCache.isXmpDevelopDownloadState( 
											readyToSaveState ) and 'xmp_develop' ) or 
			( PixelProviderSharedBlobCache.isXmpSidecarDownloadState(
											readyToSaveState ) and 'xmp_sidecar' ) or
			nil

	local ozBinarySubtype = 
			( ozBinaryType == 'xmp_develop' and 'xmp_develop' ) or
			( ozBinaryType == 'xmp_sidecar' and 'xmp_sidecar' ) or
			( ozBinaryType == 'original' and 'master' ) or
			( ozBinaryType == 'proxy' and 
				PixelProviderSharedTaskUtils_maybeGetBinarySubtypeForProxy( 
					baseRenderInfoCache, localAssetId, preferredSize ) )
	
	if not ozBinarySubtype then

		logger:errorf( "No valid blob type for state %s type %s in PixelProviderSaveDownloadedTask for %s.%s.", 
							readyToSaveState, ozBinaryType, pixStreamId,
							localAssetId )

		scheduler:_applyTransitionToRequest( needToSaveRenderBinaryRequest,
									'no_binary_blob' )

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

		return

	end

	local binaryBlob, maybeBlobSha256

	if ozBinaryType == 'original' then

		binaryBlob, maybeBlobSha256 = 
			pendingSaveBlobCache:popOriginalBlob( localAssetId )

	elseif ozBinaryType == 'xmp_develop' then

		binaryBlob, maybeBlobSha256 = 
			pendingSaveBlobCache:popXmpDevelopBlob( localAssetId )

	else

		binaryBlob = pendingSaveBlobCache:popBlob( localAssetId,
														ozBinaryType,
														ozBinarySubtype )

	end

	-- now update all needToSaveRenderBinaryStates requests that can
	-- also be fulfilled with this blob. In practice, this is almost always
	-- a single request, but it may also include pending downloads

	local getRequestsToUpdate = 
				binaryBlob and 
					PixelProviderSharedBlobCache.getCanJoinNegativeSaveRequests or
					PixelProviderSharedBlobCache.getNeedNegativeSaveRequests

	local requestsToApplyBeginSaveTransition = 
			getRequestsToUpdate( scheduler, localAssetId,
										ozBinaryType )

	local beginSaveTransition = binaryBlob and 'start_save_download' or 'no_binary_blob'

	-- update the states
	scheduler:_applyTransitionAllRequests( 
					requestsToApplyBeginSaveTransition,
					beginSaveTransition )

	-- (maybe) start saving a download
	if not binaryBlob then

		WFFunctionUtils_map( sendTestMessage, 
								requestsToApplyBeginSaveTransition,
								scheduler, 'skipSaveNoBlob' )

		return

	end

	if #requestsToApplyBeginSaveTransition == 0 then

		binaryBlob:close()

		binaryBlob = nil

		return

	end

	return saveLocalBinaryResultStream( taskState.docStore,
											taskState.fileSpecObj,
											taskState.saveToDiskThread,
											localAssetId, ozBinaryType,
											ozBinarySubtype, binaryBlob, 
											maybeBlobSha256 )
		:subscribe(
			function( fileSavePath )
				
				taskState.saveResultsStream
					:onNext( 
						true, 
						fileSavePath, 
						pendingSaveNumber, 
						ozBinaryType,
						ozBinarySubtype, 
						binaryBlob, 
						localAssetId )

			end,
			function( errStr )

				taskState.saveResultsStream
					:onNext( 
						false, 
						errStr,
						pendingSaveNumber,
						ozBinaryType,
						ozBinarySubtype,
						binaryBlob,
						localAssetId )

			end )

end

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

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

	local localAssetIdsWithBlob = pendingSaveBlobCache:localAssetIdsWithBlob()

	-- first purge stale items from the blob cache
	purgeAbandonedBlobs( scheduler, pendingSaveBlobCache,
							localAssetIdsWithBlob )

	-- now see if we have any requests that will be fulfilled when the active save
	-- completes. For those items, join the active save
	joinRequestsToActiveSaveRequests( scheduler, localAssetIdsWithBlob )

	-- now try to do some saves

	local pendingSaveNumber

	repeat

		pendingSaveNumber = maybeGetUnusedPendingSaveNumber(
									taskState )

		if not pendingSaveNumber then

			break

		end

		local needToSaveRenderBinaryRequest =
					{ PixelPathProviderSchedulerPrivateMethods.getNextRequestWithState( 
						scheduler,  
						unpack( PixelProviderSharedBlobCache.needToSaveRenderBinaryStates ) ) }

		if not next( needToSaveRenderBinaryRequest ) then

			break

		end

		taskState.pendingSaveSubscriptions[ pendingSaveNumber ] = 
			updateRequestAndMaybeSubscribeToSave( 
				taskState, scheduler, baseRenderInfoCache,
				pendingSaveBlobCache, pendingSaveNumber,
				needToSaveRenderBinaryRequest )

	until false

end

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

local function cleanup( taskState )

	
	for _, subscription in ipairs( taskState.pendingSaveSubscriptions ) do

		subscription:dispose()

	end

	taskState.pendingSaveSubscriptions = nil

	taskState.saveResultsStream:dispose()

	taskState.saveToDiskThread:close()

	taskState.saveToDiskThread = nil

end

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

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup,

	_privateTestExport = {

		TREAT_PERSISTENT_XMP_DEVELOP_SHA256_MISMATCH_AS_FATAL = 
			TREAT_PERSISTENT_XMP_DEVELOP_SHA256_MISMATCH_AS_FATAL,

		getXmpDevelopSavePath = getXmpDevelopSavePath,

	}

}
