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

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.oz_client.managed_catalog
local WFCatFileSpec = require 'WFCatFileSpec'

	-- WF.core.utils
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 KSDate = require 'KSDate'
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
local getBinarySubtypeForBinaryType = PixelProviderSharedTaskUtils.getBinarySubtypeForBinaryType

--------------------------------------------------------------------------------
-- 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 function handleSavedBlob( pendingSaveSubscriptions, scheduler, 
									localBinaryPathCache,
									pendingSaveBlobCache,
									ok, maybeErr, pendingSaveNumber,
									ozBinaryType, ozBinarySizeName,
									binaryBlob, localAssetId,
									filePath )

	pendingSaveSubscriptions[ pendingSaveNumber ]:dispose()

	pendingSaveSubscriptions[ pendingSaveNumber ] = nil

	if maybeErr then

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

	else

		logger:infof( "Did save %s %s for localAssetId %s at %s",
									ozBinaryType, ozBinarySizeName,
									localAssetId, filePath )

	end

	local getRequestsToUpdate = 
				ok and 
					PixelProviderSharedBlobCache.getFulfilledBySuccessfulNegativeSaveRequests or
					PixelProviderSharedBlobCache.getWaitingForNegativeSaveRequests

	local requestsToApplyBinarySaveTransition = 
			getRequestsToUpdate( 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 )

		if binaryBlob then

			binaryBlob:close()

		end

		return

	end
		
	if ok then

		binaryBlob:close()

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

	else

		-- push the blob back into the cache so we can try again

		pendingSaveBlobCache:pushBlob( localAssetId, binaryBlob, ozBinaryType,
										ozBinarySizeName )

		binaryBlob = nil

	end

	local binarySaveTransition = ok and 'saved_valid_download' or 'retry_save_download'

	scheduler:_applyTransitionAllRequests( 
					requestsToApplyBinarySaveTransition,
					binarySaveTransition )

	local testMessage = ok and 'didPixelCacheSave' or 'saveFailed'

	WFFunctionUtils_map( sendTestMessage, 
							requestsToApplyBinarySaveTransition, scheduler,
							testMessage )

end

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

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

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

local function setupTask( scheduler, docStore, fileSpecObj, 
							localBinaryPathCache,
							pendingSaveBlobCache, 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, maybeErr, pendingSaveNumber, ozBinaryType,
						ozBinarySizeName, binaryBlob, localAssetId, filePath )

				local ok2, maybeError =
						pcall( handleSavedBlob, 
								pendingSaveSubscriptions, scheduler,
								localBinaryPathCache, pendingSaveBlobCache, ok,
								maybeErr, pendingSaveNumber, ozBinaryType,
								ozBinarySizeName, binaryBlob, localAssetId,
								filePath )

				if not ok2 then

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

				end

			end,
			
			function( errStr )

				logger:errorf( "error in finishedSaveSubscription: %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 generateLocalFilePathForOriginal( fileSpecObj, docValue )

	local captureDate = docValue.captureDate
	
	captureDate = type( captureDate ) == 'string' and 
					KSDate.isKSDate( captureDate ) and
					captureDate or 
					'0000-00-00T00:00:00.000' -- Unknown date
	
	local fileName = derefOpt( docValue, 'importSource', 'fileName' )

	fileName = type( fileName ) == "string" and
					fileName or KSIdGenerator.uuidInLowercaseHex()

	return fileSpecObj:pathForOriginal( fileName, captureDate, true )

end

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

local function generateLocalFilePathForProxy( fileSpecObj, docValue, 
												ozBinarySizeName )

	local assetId = derefOpt( docValue, '_id' )

	return assetId and fileSpecObj:pathForProxy( assetId, ozBinarySizeName )

end

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

local function generateLocalFilePathForXmpDevelop( fileSpecObj, docValue )

	-- SPAM [matthewj 2016-09-07]: This may present an issue if we somehow download
	-- a /rels/xmp/develop who's sha256 does not match the xmpCameraRaw section
	-- due to either invalid data on Oz or a race condition. In the case of invalid
	-- data we probably want to save it with the sha256 that's in the doc value 
	-- to avoid possibly creating a looper of constantly trying to download the 'correct'
	-- sha256 data which we end up never getting.
	local xmpCameraRawSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )
	
	return xmpCameraRawSha256 and fileSpecObj:pathForSettingsXmp( xmpCameraRawSha256 )
	
end

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

local function generateLocalFilePathForXmpSidecar( fileSpecObj, docValue )

	local assetId = derefOpt( docValue, '_id' )

	return assetId and fileSpecObj:pathForXmpSidecar( assetId )

end

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

local function localBinaryPathStream( docStore, fileSpecObj, localAssetId,
											ozBinaryType, ozBinarySizeName )

	return docStore
			:docValue( localAssetId )
			:map( 

				function( docValue )

					if docValue._empty then
						local errStr = string.format( "Binary type %s for localAssetId %s has been deleted",
														ozBinaryType, localAssetId )
						error( errStr )
					end

					if ozBinaryType ~= 'original' and ozBinaryType ~= 'proxy' and
							ozBinaryType ~= 'xmp_develop' and ozBinaryType ~= 'xmp_sidecar' then
						
						local errStr = string.format( "Binary type not supported by PixelProviderSaveDownloadedTask: %s", ozBinaryType )
						error( errStr )
					
					end
					
					local filePath 
					local sha256
					
					if ozBinaryType == 'original' then
						filePath = generateLocalFilePathForOriginal( fileSpecObj, docValue )
						local _, isInvalid = WFCatFileSpec.getOriginalLinkFromDocValue( docValue )
						sha256 = isInvalid and 'invalid' or derefOpt( docValue, 'importSource', 'sha256' )
					elseif ozBinaryType == 'proxy' then
						filePath = generateLocalFilePathForProxy( fileSpecObj, docValue, ozBinarySizeName )
						local _, _, isInvalid = WFCatFileSpec.getProxyLinkFromDocValue( docValue )
						sha256 = isInvalid and 'invalid' or derefOpt( docValue, 'importSource', 'sha256' )
					elseif ozBinaryType == 'xmp_develop' then
						filePath = generateLocalFilePathForXmpDevelop( fileSpecObj, docValue )
					elseif ozBinaryType == 'xmp_sidecar' then
						filePath = generateLocalFilePathForXmpSidecar( fileSpecObj, docValue )
					end
					
					if not filePath then
					
						local errStr = string.format( "Unable to create a %s file path for localAssetId %s",
															ozBinaryType, localAssetId )
						error( errStr )
					
					end
					
					if ozBinaryType == 'original' or ozBinaryType == 'proxy' then
					
						-- Ensure that the path is unique

						filePath = WFFileUtils.ensureFileNameIsUnique( filePath )
					
					end
					
					return filePath, sha256

				end )

end

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

local function writeNegativeToDiskStream( 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 moveNegativeToFinalLocation( tempPath, finalPath, localAssetId, ozBinaryType )

	logger:infof( "Now moving %s blob from %s to %s save for localAssetId %s",
													ozBinaryType, tempPath, finalPath, localAssetId )
	
	return WFAsyncFileUtils.moveToUniquelyNamedFile( tempPath, finalPath )
	
		:catch( function( err )
		
			return Rx.Observable.just( false, err )
		
		end )
	
end

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

local function updateDocStoreNegativeInfoStreamToBeforeSaveState( docStore, localAssetId,
													filePath, fileSize,
													ozBinaryType, ozBinarySizeName,
													sha256 )
													
	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
			-- getXmpDevelopLinkFromDocValue 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

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

-- finalizeDocStoreAnnotationStream 
-- 
-- 	This function has 2 intents
-- 	
-- 	1) (occurs always) Cause the docValue._changeCounter to increment so that
-- 	   both
-- 	   	
-- 	   	docValue
-- 				:ongoingDocValue()
-- 				
-- 			and
-- 			
-- 		docStore
-- 			:scannedFilteredMappedChangesFeed()
-- 				
-- 	   will emit a new docValue for localAssetId
-- 		
-- 	2) (occurs more rarely, usually when external storage present) make sure
-- 	   the docStore contains the actual final path written. Sometimes due
-- 	   to duplicate filenames, the final write path can differ from the before
-- 	   write path.
-- 	
--------------------------------------------------------------------------------

local function finalizeDocStoreAnnotationStream( 
				docStore, localAssetId, ozBinaryType, ozBinarySizeName,
				finalPath )
	
	local perBinaryPostSaveLocalOnlyData = {}

	if finalPath then
	
		local finalVolumeGuid = WFFileUtils.getVolumeGuidForPath( finalPath )

		local finalRelativeFilePath = WFFileUtils.removeVolumeFromPath( finalPath )

		if type( finalVolumeGuid ) == 'string' and 
				type( finalRelativeFilePath ) == 'string' then

			-- always set the perBinaryPostSaveLocalOnlyData path components
			-- based on finalPath. It's possible for those path components to
			-- be different from what they were before the save due to in use
			-- filename disambiguation. If they don't change, no harm in setting
			-- them to the same value as in the docStore.
			-- 
			perBinaryPostSaveLocalOnlyData.volumeGuid = finalVolumeGuid

			perBinaryPostSaveLocalOnlyData.relativePath =
				finalRelativeFilePath

		end
		
	end

	-- The lastSaveTimestamp field is intended to be uniquish every time we
	-- do a save for a binary. So, we use "now" as the value. This guarantees
	-- that perBinaryPostSaveLocalOnlyData will contain at least one value that
	-- is different from the cuurent docstore values and therefore the
	-- applyOverlay( ... ) call will update the docValue._changeCounter for the
	-- localAssetId
	perBinaryPostSaveLocalOnlyData.lastSaveTimestamp =
		KSDate.timeInIso8601Format()
	
	local perBinaryLocalOnlyAnnotationUpdateOverlay = { 

		_id = localAssetId,

		_localOnly = {
			files = {
				[ ozBinaryType ] = {
					[ tostring( ozBinarySizeName ) ] = 
						perBinaryPostSaveLocalOnlyData,
				}
			}
		}
	}
	
	return docStore
			-- we expect this :applyOverlay to always cause the
			-- docValue._changeCounter to change. If it does not, the UI client
			-- may hang in loupe view because it will not receive notifications
			-- that a new negative is available locally.
			:applyOverlay( perBinaryLocalOnlyAnnotationUpdateOverlay ) 
			:take( 1 )
		
end

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

local function updateDocStoreXmpDevelopInfoStream( docStore, localAssetId, 
							filePath, fileSize,
							binaryBlob )
	
	return WFCrypto.asyncDigestData( binaryBlob, 'SHA256' )
		:map( function( sha256 ) 
		
			local localOnly = {
				files = {
					xmp_develop = {
						path = filePath,
						fileSize = fileSize,
						sha256 = sha256,
					}
				}
			}

			local overlay = { 
				_id = localAssetId,
				_localOnly = localOnly,
			}					
			
			return overlay 
			
		end )
		:flatMap( function( overlay ) 
		
			return docStore
					:applyOverlay( overlay )
					:take( 1 )

		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 docStoreLocalBinaryPathStream( docStore, fileSpecObj,
													saveToDiskThread,
													localAssetId, ozBinaryType,
													ozBinarySizeName,
													binaryBlob )

	local localPathStream = localBinaryPathStream( docStore, fileSpecObj, 
														localAssetId,
														ozBinaryType, 
														ozBinarySizeName )
														
	-- Annotate the doc store with the existence of the negative.

	local maybeUpdatedDocstoreStream =
			localPathStream
				:flatMapLatest( function( finalPath, maybeSha256 )
				
					-- 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 finalPathVolumeGuid = finalPath and WFFileUtils.getVolumeGuidForPath( finalPath )
					
					local tempFolder = fileSpecObj:tempFolderPathForVolume( finalPathVolumeGuid )
					
					local tempFilePath = KSPathUtils.child( tempFolder, KSIdGenerator.uuidInLowercaseHex() )
					
					-- Write the negative to a temporary location on disk.

					local maybeTempPathStream = writeNegativeToDiskStream( saveToDiskThread,
																		binaryBlob, tempFilePath,
																		localAssetId,
																		ozBinaryType )
																		
					return maybeTempPathStream
						:map( function( ... )
							return finalPath, maybeSha256, ...
						end )
				
				end )
				
				:flatMapLatest( function( finalPath, maybeSha256, haveTempPath, tempPathOrErr, fileSize )

					if not haveTempPath then

						error( tempPathOrErr )

					end
					
					local updateDocStore
					
					if ozBinaryType == 'xmp_develop' then
						
						updateDocStore = updateDocStoreXmpDevelopInfoStream( 
							docStore, localAssetId, finalPath, fileSize,
							binaryBlob )

					elseif ozBinaryType == 'xmp_sidecar' then

						updateDocStore = updateDocStoreXmpSidecarInfoStream(
							docStore, localAssetId, finalPath, binaryBlob )

					else 
						updateDocStore = updateDocStoreNegativeInfoStreamToBeforeSaveState( 
							docStore, localAssetId, finalPath, fileSize,
							ozBinaryType, ozBinarySizeName, maybeSha256 )
					end
					
					return updateDocStore
						:map( function()
							return tempPathOrErr, finalPath
						end )
						
				end )

	-- Move the negative to the final location.
	-- SPAM [jandrew 2016-11-30]: 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.

	local maybeMoveNegative = 
			maybeUpdatedDocstoreStream
				:flatMapLatest( function( tempPath, finalPath )
					return moveNegativeToFinalLocation( tempPath, finalPath, localAssetId, ozBinaryType )
				end )
				
	-- Finalize the doc store annotation. 
	-- 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. 
				
	local finalizeDocStoreAnnotation = 
			maybeMoveNegative
				:flatMapLatest( function( finalPath )
					return finalizeDocStoreAnnotationStream( docStore, localAssetId, ozBinaryType, ozBinarySizeName, finalPath )
						:map( function()
							return finalPath
						end )
				end )
				
	return finalizeDocStoreAnnotation
	
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 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 saveToDiskThread = taskState.saveToDiskThread

	local saveResultsStream = taskState.saveResultsStream

	local pendingSaveNumber

	repeat

		pendingSaveNumber = pendingSaveNumber or 
								maybeGetUnusedPendingSaveNumber(
									taskState )

		if not pendingSaveNumber then

			break

		end

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

		local pixStreamId, localAssetId, _, _, readyToSaveState = unpack( request )

		if not pixStreamId then

			break

		end

		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
		
		if ozBinaryType ~= 'original' and ozBinaryType ~= 'proxy'
			and ozBinaryType ~= 'xmp_develop' and ozBinaryType ~= 'xmp_sidecar' then

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

			scheduler:_applyTransitionToRequest( request,
										'no_binary_blob' )

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

		else
	
			local ozBinarySubtype = getBinarySubtypeForBinaryType(
																localAssetId,
																ozBinaryType,
																baseRenderInfoCache )

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

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

			elseif #requestsToApplyBeginSaveTransition == 0 then

				binaryBlob:close()

				binaryBlob = nil

			else

				local currPendingSaveNumber = pendingSaveNumber

				pendingSaveNumber = nil

				logger:infof( "About to save %s %s for localAssetId %s",
								ozBinaryType, ozBinarySubtype, localAssetId )

				taskState.pendingSaveSubscriptions[ currPendingSaveNumber ] = 
					
					docStoreLocalBinaryPathStream( taskState.docStore,
														taskState.fileSpecObj,
														saveToDiskThread,
														localAssetId, ozBinaryType,
														ozBinarySubtype, binaryBlob )
					:subscribe(
						function( filePath )
							
							saveResultsStream:onNext( true, nil, 
														currPendingSaveNumber,
														ozBinaryType,
														ozBinarySubtype,
														binaryBlob,
														localAssetId, 
														filePath )

						end,
						function( errStr )

							saveResultsStream:onNext( false, errStr,
														currPendingSaveNumber,
														ozBinaryType,
														ozBinarySubtype,
														binaryBlob,
														localAssetId )

						end )

			end

		end

	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,

}
