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

WFCatIdleDownloader.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.

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

	-- RxLua
local Rx = require 'Rx'
local BehaviorSubject = Rx.BehaviorSubject
local ReplaySubject = Rx.ReplaySubject
local Observable = Rx.Observable

	-- WF.core.logger
local WFLogger = require 'WFLogger'

	-- WF.core.utils
local KSData = require 'KSData'
local KSDate = require 'KSDate'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxLowFidelityTimer = require 'WFRxLowFidelityTimer'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.oz_client.managed_catalog
local WFCatDeniedListHelper = require 'WFCatDeniedListHelper'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatPhotoImportance = require 'WFCatPhotoImportance'
local WFCatSyncStatus = require 'WFCatSyncStatus'

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

local check = assert( WFArguments.check )
KSData.registerArgumentChecker()

local derefOpt = KSTableUtils.derefOpt

local runUntilDone = WFRxRunUntilDone

local KSDate_timeNow = KSDate.timeNow

local isValidAssetDocument = assert( WFCatIndexUtils.isValidAssetDocument )
local isValidVersionDocument = assert( WFCatIndexUtils.isValidVersionDocument )

local logger = WFLogger 'WFCatIdleDownloader'

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

local maxScheduledIdleDownloads = 20
	-- SPAM [jandrew 2016-06-09]: The maximum number of downloads to schedule at once. 
	-- The WFRxRateLimiter used in the HTTP implementation limits the number of concurrent
	-- HTTP requests to 5. However, it's good to queue up more than the max number of 
	-- concurrent requests so that we can quickly process the next batch without needing
	-- to wait for the next batch to be scheduled.
	
local downloadBlackoutWindowInSeconds = 60
	-- Number of seconds we disable the redownloading of a binary after a 
	-- successful download.
	--
 	-- SPAM [jandrew 05/24/2016]: Increasing this from 20 to 60 removed all the 
	-- duplicate downloads with a catalog of 10,000 assets. Duplicate downloads
	-- were occuring when a download finished, but the index store failed to update
	-- within downloadBlackoutWindowInSeconds.
	
local numberOfTriesBeforeBlockingBinary = 2
	-- Number of attempts to download a binary before blocking the binary from sync.

local failedDownloadBlackoutWindowInSeconds = 60
	-- Number of seconds a binary remains blocked after it has failed to download.

local triggerDownloadAttemptEveryNSeconds = 300
	-- Attempt to trigger the idle downloader every N seconds, as a last resort. This is only
	-- used as a way to work around sync stalls. We're not sure what's causing the sync stalls,
	-- but forcing the idle downloader to try to download every N seconds might resolve them.

local maximumWaitForDownloadWindowInSeconds = 1800
	-- After we call the download function for a binary, wait up to N seconds for any sort of response. If
	-- no response is received, we assume the download command was lost and allow it to be reissued. 

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

local function distinctTableCompare( _, tbl1, _, tbl2 )

	return KSTableUtils.deepEquals( tbl1, tbl2 )

end

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

local function clearWaitingForProxyStatus( docStore, assetId )

	return docStore:applyOverlay(
		{
			_id = assetId,
			_serverProvidedMetadata = {
				startedWaitingForProxyTimestamp = { _delete = true },
			},
		} )
		:call( runUntilDone )
		
end

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

local function docValueHasLocalOriginal( docValue )

	local localOriginalVolumeGuid = WFCatFileSpec.getOriginalVolumeGuidFromDocValue( docValue )
	local localOriginalRelativePath = WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )
	return localOriginalVolumeGuid and localOriginalRelativePath

end

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

local function docValueHasLocalProxy( docValue )

	local localProxyVolumeGuid = WFCatFileSpec.getProxyVolumeGuidFromDocValue( docValue )
	local localProxyRelativePath = WFCatFileSpec.getProxyRelativePathFromDocValue( docValue )
	return localProxyVolumeGuid and localProxyRelativePath

end

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

local function addWaitingForProxyAnnotation( self, wfCatMetadataPuller, result )

	local hasLocalOriginal = docValueHasLocalOriginal( result )
	local hasLocalProxy = docValueHasLocalProxy( result )

	local hasOriginal = WFCatFileSpec.getOriginalLinkFromDocValue( result )
	local hasProxy = WFCatFileSpec.getProxyLinkFromDocValue( result )
	
	-- Only add the annotation when we don't have a local copy of the original or
	-- proxy, and when Oz only has the original.
	
	if not hasLocalOriginal and hasOriginal and
		not hasLocalProxy and not hasProxy then
	
		local lastUpdated = KSDate.timeInIso8601Format()
	
		-- SPAM: There can exist a slight delay between the time it takes for an asset to show
		-- up on Oz and when Oz generates a proxy for said asset. Many of the Nimbus
		-- systems default to syncing and using the original if the proxy isn't available
		-- on Oz. In order to prevent us from being too agressive with using originals, we
		-- annotate the doc store with `startedWaitingForProxyTimestamp` which helps us determine
		-- if we should wait a bit longer for the proxy to appear on Oz, or if we should
		-- go ahead and use the original.

		result._serverProvidedMetadata.startedWaitingForProxyTimestamp = lastUpdated

	end

end	

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

local function checkIfMaybeDenied( docValue )

	local deniedListKey = 'WFCatIdleDownloader'

	local isDenied = WFCatDeniedListHelper.isDeniedForKey( docValue, deniedListKey )
	
	return isDenied and 1 or 0
	
end
	
--------------------------------------------------------------------------------

local function hasLocalMasterOrNeedsDownload( docValue )

	local hasLocal = docValueHasLocalOriginal( docValue )
	
	local ozDownloadUrl = WFCatFileSpec.getOriginalLinkFromDocValue( docValue ) ~= nil
												
	local needsDownload = ( ozDownloadUrl and not hasLocal ) and 1 or 0
	
	hasLocal = hasLocal and 1 or 0
	
	ozDownloadUrl = ozDownloadUrl and 1 or 0
	
	return needsDownload, hasLocal, ozDownloadUrl

end

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

local function hasLocalXmpDevelopOrNeedsDownload( docValue )

	local hasLocal = WFCatFileSpec.getXmpDevelopInfoFromDocValue( docValue ) ~= nil

	local ozDownloadUrl = WFCatFileSpec.getXmpDevelopLinkFromDocValue( docValue ) ~= nil
												
	local needsDownload = ( ozDownloadUrl and not hasLocal ) and 1 or 0

	return needsDownload

end

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

local function dateIsWithinNSeconds( date, seconds )

	local nDaysAgoInSeconds = KSDate_timeNow() - seconds

	local ok, dateInSeconds = pcall( KSDate.iso8601ToTimeIntervalSinceReference, date )
	
	if not ok then 
		logger:warnf( "failed to parse date %s: %s", date, dateInSeconds )
		return false
	end
	
	return dateInSeconds >= nDaysAgoInSeconds
	
end

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

local function hasLocalProxyOrNeedsDownload( docValue )

	local maybeWaitingForProxyOnOzTimestamp = derefOpt( docValue, '_serverProvidedMetadata', 'startedWaitingForProxyTimestamp' )

	local hasLocalProxy = docValueHasLocalProxy( docValue )
	
	local ozDownloadUrl, proxySize = WFCatFileSpec.getProxyLinkFromDocValue( docValue )

	local waitingForProxyOnOz = maybeWaitingForProxyOnOzTimestamp
		
	local needsDownload = 0
	
	if not hasLocalProxy then

		if waitingForProxyOnOz or ozDownloadUrl then
			needsDownload = 1
		end

	end
	
	hasLocalProxy = hasLocalProxy and 1 or 0
	
	ozDownloadUrl = ozDownloadUrl and 1 or 0
	
	waitingForProxyOnOz = waitingForProxyOnOz or 0
		-- Emitting this as 0 instead of nil because SQL `waitingForProxyOnOz > 0` is 
		-- faster than `waitingForProxyOnOz != NULL`.

	return needsDownload, hasLocalProxy, ozDownloadUrl, proxySize, waitingForProxyOnOz

end

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

local function getDatesFromDocValue( docValue )

	local importDate = derefOpt( docValue, 'importSource', 'importTimestamp' ) or '0000-00-00T00:00:00.000'
	
	local captureDate = derefOpt( docValue, 'captureDate' ) or '0000-00-00T00:00:00.000'
	
	return importDate, captureDate

end

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

local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )
	
	indexStoreSchema:addTable {

		sql = [[CREATE TABLE localStatusAndOzStatus(
				needsOriginalDownload,
				hasLocalOriginal,
				ozHasOriginal,
				needsProxyDownload,
				hasLocalProxy,
				ozHasProxy,
				proxySize,
				needsDevelopXmpDownload,
				importDate,
				captureDate,
				waitingForProxyOnOz,
				isMaybeDenied,
				isImportant,
				isMarkedForOffline,
				ozHasNonThumbnailRendition,
				subtype,
				oldIsMarkedForOffline,
				isVersionDocument )]],

		useLocalId = true,
		
		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ 'asset', '*' },
			{ 'version', '*' },
		},
		
		map = function( docValue, emit )

			local emitFnVersion = 3
				-- Increment this version to trigger reindexing of this table when the implementation of 
				-- any of the fns called within change. This value is passed to isValidAssetDocument
				-- where it is ignored. We do this to remove any unused local warnings from Halide.
				
			if isValidAssetDocument( docValue, emitFnVersion ) then

				local needsOriginalDownload, hasLocalOriginal, ozHasOriginal 
						= hasLocalMasterOrNeedsDownload( docValue )
				
				local needsProxyDownload, hasLocalProxy, ozHasProxy, proxySize, waitingForProxyOnOz 
						= hasLocalProxyOrNeedsDownload( docValue )

				local needsDevelopXmpDownload = hasLocalXmpDevelopOrNeedsDownload( docValue )

				local importDate, captureDate = getDatesFromDocValue( docValue )

				local isMaybeDenied = checkIfMaybeDenied( docValue )

				local isImportant = WFCatPhotoImportance.isPhotoImportant( docValue ) and 1 or 0

				local isMarkedForOffline = WFCatFileSpec.getMarkedForOfflineUseFromDocValue( docValue ) and 1 or 0

				local oldIsMarkedForOffline = derefOpt( docValue, '_localOnly', 'markedForOfflineUse' ) and 1 or 0

				local ozHasNonThumbnailRendition = WFCatFileSpec.checkOzForNonThumbnailRendition( docValue ) and 1 or 0
				
				local subtype = docValue.subtype
				
				local isVersionDocument = 0
				
				emit( needsOriginalDownload, hasLocalOriginal, ozHasOriginal, 
						needsProxyDownload, hasLocalProxy, ozHasProxy, proxySize,
						needsDevelopXmpDownload, importDate, captureDate, waitingForProxyOnOz, 
						isMaybeDenied, isImportant, isMarkedForOffline,
						ozHasNonThumbnailRendition, subtype, oldIsMarkedForOffline, isVersionDocument )
						
			elseif isValidVersionDocument( docValue, emitFnVersion ) then
			
				-- START: The following don't apply to versions.
				
				local needsOriginalDownload = 0
				local hasLocalOriginal = 0
				local ozHasOriginal = 0
				local needsProxyDownload = 0
				local hasLocalProxy = 0
				local ozHasProxy = 0
				local proxySize = 0
				local waitingForProxyOnOz = 0
				local oldIsMarkedForOffline = 0
				local isMarkedForOffline = 0
				local ozHasNonThumbnailRendition = 0
				
				-- END
				
				local importDate = derefOpt( docValue, 'userCreated' ) 
				local captureDate = importDate
					-- Version documents don't have import or capture dates, so instead use the
					-- creation date for both.

				local needsDevelopXmpDownload = hasLocalXmpDevelopOrNeedsDownload( docValue )

				local isMaybeDenied = checkIfMaybeDenied( docValue )

				local isImportant = WFCatPhotoImportance.isPhotoImportant( docValue ) and 1 or 0

				local subtype = docValue.subtype
				
				local isVersionDocument = 1
				
				emit( needsOriginalDownload, hasLocalOriginal, ozHasOriginal, 
						needsProxyDownload, hasLocalProxy, ozHasProxy, proxySize,
						needsDevelopXmpDownload, importDate, captureDate, waitingForProxyOnOz, 
						isMaybeDenied, isImportant, isMarkedForOffline,
						ozHasNonThumbnailRendition, subtype, oldIsMarkedForOffline, isVersionDocument )
			
			end

		end,

	}
	
	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_importantOriginal
			ON localStatusAndOzStatus( 
				isImportant, 
				hasLocalOriginal )]]

	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_importantProxy
			ON localStatusAndOzStatus( 
				isImportant, 
				hasLocalProxy )]]
				
	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_proxy
			ON localStatusAndOzStatus( 
				needsProxyDownload, 
				waitingForProxyOnOz )]]
	
	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_original
			ON localStatusAndOzStatus( 
				hasLocalProxy, 
				needsOriginalDownload )]]

	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_offline
			ON localStatusAndOzStatus( 
				needsOriginalDownload, 
				isMarkedForOffline )]]
				
	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_xmpDevelop
			ON localStatusAndOzStatus( needsDevelopXmpDownload )]]
			
	indexStoreSchema:addIndex
		[[CREATE INDEX localStatusAndOzStatus_waitingForProxyOnOz
			ON localStatusAndOzStatus( waitingForProxyOnOz )]]
				
end

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

local function addSearchToSearchSchema( WFCatIdleDownloader, searchSchema )

	searchSchema:addSearchableData {
		sql_table = 'localStatusAndOzStatus',
		column = 'needsOriginalDownload',
		dataType = 'uint',
		term = 'needsToDownloadOriginal',
	}

	searchSchema:addSearchableData {
		sql_table = 'localStatusAndOzStatus',
		column = 'needsProxyDownload',
		dataType = 'uint',
		term = 'needsToDownloadProxy',
	}

end

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

local function getCachedBinaryInfo( self, localId )

	local binaryInfo = self._binaryInfoCache[ localId ]
	
	local binaryType = derefOpt( binaryInfo, 'binaryType' )
	local binarySize = derefOpt( binaryInfo, 'binarySize' )
	local isMarkedForOffline = derefOpt( binaryInfo, 'isMarkedForOffline' )
	local hasLocalOriginal = derefOpt( binaryInfo, 'hasLocalOriginal' )
	
	return binaryType, binarySize, isMarkedForOffline, hasLocalOriginal
	
end

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

local function createRecentlyDownloadedKey( localId, binaryType )
	if localId and binaryType then
		return localId .. "::" .. binaryType
	end
end

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

local function maybeRemoveDownloadRequiredStatus( self, localId, bType, downloadRequired )

	if not downloadRequired then
	
		local key = createRecentlyDownloadedKey( localId, bType )
		self._requiredDownloadsInFlight[ key ] = nil
		
	end
	
end

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

local function cacheBinaryInfo( self, localId, bType, bSize, isMarkedForOffline, hasLocalOriginal, downloadRequired )

	if downloadRequired then
		local key = createRecentlyDownloadedKey( localId, bType )
		self._requiredDownloadsInFlight[ key ] = true
		self._isDownloadingRequiredBinaryStream:onNext( true )
	end

	self._binaryInfoCache[ localId ] = {
		binaryType = bType,
		binarySize = bSize,
		isMarkedForOffline = isMarkedForOffline,
		hasLocalOriginal = hasLocalOriginal,
		downloadRequired = downloadRequired
	}
	
end

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

local function clearBinaryInfoCache( self, localId )

	local binaryInfo = self._binaryInfoCache[ localId ]
	local downloadRequired = derefOpt( binaryInfo, 'downloadRequired' )
	
	if downloadRequired then
		local bType = derefOpt( binaryInfo, 'binaryType' )
		local key = createRecentlyDownloadedKey( localId, bType )
		self._requiredDownloadsInFlight[ key ] = nil
	end
	
	if next( self._requiredDownloadsInFlight ) == nil then
	
		self._isDownloadingRequiredBinaryStream:onNext( false )
		
	end
	
	self._binaryInfoCache[ localId ] = nil

end

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

local function clearRecentlyDownloaded( self, localId, binaryType )

	local key = createRecentlyDownloadedKey( localId, binaryType )
	logger:infof( "Clearing download blackout for photo: %s", key )
	
	local maybeSubscription = self._recentlyDownloadedBinaries[ key ] 
	self._recentlyDownloadedBinaries[ key ] = nil
	
	if maybeSubscription then
		maybeSubscription:dispose()
	end

end

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

local function trackDownloadForDiagnostics( self, binaryType )

	local function updateDocumentCount( typeOfBinary )

		local total = derefOpt( self._binariesDownloadedByType, typeOfBinary ) or 0

		KSTableUtils.setNested( self._binariesDownloadedByType, total + 1, typeOfBinary )

		return self._binariesDownloadedByType

	end

	self._binariesDownloadedByType = updateDocumentCount( binaryType )

	if self._binariesDownloadedByTypeStream then
		self._binariesDownloadedByTypeStream:onNext( self._binariesDownloadedByType )
	end

end

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

local function markAsRecentlyDownloaded( self, localId, binaryType )

	local key = createRecentlyDownloadedKey( localId, binaryType )
	logger:infof( "Creating blackout window for photo: %s", key )
	
	-- After we call the download function for a binary, we mark is as recently downloaded.
	-- We only clear the recently downloaded flag when there's a response on the download stream
	-- from the pixel provider. If after N seconds, no response is received, we assume the PP
	-- dropped the request and we clear the recently downloaded status; this allows the request
	-- to be tried again at a later time.
	
	trackDownloadForDiagnostics( self, binaryType )
	
	self._recentlyDownloadedBinaries[ key ] = 
			WFRxLowFidelityTimer.timer( maximumWaitForDownloadWindowInSeconds )
			:tap( function()
				-- TODO: Cancel the existing download request, if present. 
				clearRecentlyDownloaded( self, localId, binaryType )
			end )
			:subscribe()

end

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

local function wasRecentlyDownloaded( self, localId, binaryType )

	local key = createRecentlyDownloadedKey( localId, binaryType )
	return self._recentlyDownloadedBinaries[ key ] and true

end

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

local function doClearBlackoutWindow( self, localId, binaryType )

	-- Because the index store takes a few seconds to update, and the doc store
	-- could have potentially just updated, we will give the index store some time
	-- to bring itself up to date.
	
	-- TODO: Is there a more deterministic way to do this? An index store API that fires
	-- when it's up to date?

	local indexStoreUpdateBuffer = 30
	
	WFRxLowFidelityTimer.timer( indexStoreUpdateBuffer )
		:tap( function()
			clearRecentlyDownloaded( self, localId, binaryType )
		end )
		:call( runUntilDone )
		
end

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

local function eventuallyClearBlackoutWindow( self, localId, binaryType )

	WFRxLowFidelityTimer.docValueTimer( downloadBlackoutWindowInSeconds, self._catalog:docStore(), localId )
	
		:tap( function( docValue )
															
			local hasLocalPath = docValueHasLocalOriginal( docValue )
			local fileSize = WFCatFileSpec.getOriginalSizeFromDocValue( docValue )
			
			if binaryType == 'proxy' then
				hasLocalPath = docValueHasLocalProxy( docValue ) 
				fileSize = WFCatFileSpec.getProxySizeFromDocValue( docValue )
			end
			
			if hasLocalPath or not isValidAssetDocument( docValue ) then

				local remainingSpace = self._estimatedRemainingSpace[ binaryType ]

				if remainingSpace and fileSize then
					self._estimatedRemainingSpace[ binaryType ] = remainingSpace - fileSize
				end

				-- Once the docValue contains the negative path, we can safely remove
				-- the asset from the recently downloaded table.
				
				doClearBlackoutWindow( self, localId, binaryType )
				
			else
				eventuallyClearBlackoutWindow( self, localId, binaryType )
			end
			
		end )
		
		:call( runUntilDone )

end

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

local function downloadCompletedOrDenied( self, localId, binaryType, binarySize, completed )

	self._downloadsInFlightCount = self._downloadsInFlightCount - 1
				
	clearBinaryInfoCache( self, localId )
	
	self._downloadFinished:onNext( localId )
	
	-- Prevent the binary from being redownloaded within the blackout window.
	
	if completed then
	
		-- If the download completed, than we want to wait to clear the recently-downloaded
		-- status until we've detected that the doc store is properly updated with the
		-- file existence annotation.
		
		eventuallyClearBlackoutWindow( self, localId, binaryType )
	
	else
		
		-- If the download was denied, there's no need to wait for a file existence
		-- annotation so we clear the recently-downloaded status right away.
	
		doClearBlackoutWindow( self, localId, binaryType )
		
	end
	
end

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

local function setupDownloadListener( self )

	local function handleDownloadStreamEvent( _, localId, success )
		
		local binaryType, binarySize, isMarkedForOffline, hasLocalOriginal = getCachedBinaryInfo( self, localId )
		
		if wasRecentlyDownloaded( self, localId, binaryType ) then
			
			if success then
			
				logger:infof( "Successfully downloaded binary for local Id: %s", localId )
				
			else
				
				logger:infof( "Failed to download binary for local Id: %s", localId )
			
				if not self._failedDownloadDeniedListHelper:maybeDeny( localId ) then
				
					-- Download wasn't denied, so let's try it again.
					
					if binaryType and binarySize then
						return self._downloadFunc( localId, binaryType, binarySize, isMarkedForOffline, hasLocalOriginal )
					end
					
				else
				
					WFPerformanceReport.recordSingleValue( 'WFCatIdleDownloader:DeniedList', {
						localId = localId,
						binaryType = binaryType,
						binarySize = binarySize,
						numberOfFailedAttempts = numberOfTriesBeforeBlockingBinary,
						deniedListDuration = failedDownloadBlackoutWindowInSeconds
					} )
					
				end
		
			end
		
			downloadCompletedOrDenied( self, localId, binaryType, binarySize, success )
		
		end

	end
		
	self._downloadStreamSubscription = self._downloadStream
		:subscribe( handleDownloadStreamEvent )

end

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

local function maybePerformanceWrap( id, observable )

	id = id or 'unknown'
	return WFPerformanceReport.wrapObservable( 'WFCatIdleDownloader: ' .. id, observable )

end 

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

local function abandonDownloadsOfType( self, binaryType, onlyAbandonNonRequired )

	local ids = {}

	for localId, info in pairs( self._binaryInfoCache ) do
	
		if binaryType == info.binaryType then
		
			if not onlyAbandonNonRequired or not info.downloadRequired then
			
				table.insert( ids, localId )
				downloadCompletedOrDenied( self, localId, binaryType, info.binarySize )
		
			end
		
		end
	
	end

	self._batchCancelDownloadsFunc( binaryType, ids )
	
end

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

local function abandonAllDownloadsOfType( self, binaryType )

	 abandonDownloadsOfType( self, binaryType, false )
	
end

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

local function abandonNonRequiredDownloadsOfType( self, binaryType )

	 abandonDownloadsOfType( self, binaryType, true )

end

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

--[[
	Fires an array of volumes and statuses:
	{
		'<binary type>' = {
			volumeGuid = <string>
			online = <boolean>
			outOfSpace = <boolean>
		},
		'<binary type>' = {
			volumeGuid = <string>
			online = <boolean>
			outOfSpace = <boolean>
		},
		...
	}
]]--

local function reportVolumeStatusChange( self )

	check( 'WFCatIdleDownloader' )
		
	local volumeStatuses = {}
	
	for binaryType, volumeGuid in pairs( self._volumes ) do
	
		local offline = self._statusOfVolumes.offlineVolumes[ volumeGuid ]
		local outOfSpace = self._statusOfVolumes.volumesOutOfSpace[ volumeGuid ]
		local overIdleSpaceThreshold = self._statusOfVolumes.volumesOverIdleSpaceThreshold[ volumeGuid ]
		
		volumeStatuses[ binaryType ] = {
			volumeGuid = volumeGuid,
			offline = offline,
			outOfSpace = outOfSpace,
			overIdleSpaceThreshold = overIdleSpaceThreshold
		}
		
		self._estimatedRemainingSpace[ binaryType ] = self._statusOfVolumes.remainingSpace[ volumeGuid ]
		logger:infof( "~~~~ACTUAL remaining space for %s: %s~~~~", binaryType, self._estimatedRemainingSpace[ binaryType ] )
		
	end
			
	self._statusOfVolumesStream:onNext( volumeStatuses )

end

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

local function isVolumeFullOrOffline( self, volumeGuid )

	return volumeGuid 
		and ( self._statusOfVolumes.volumesOutOfSpace[ volumeGuid ]
		or self._statusOfVolumes.offlineVolumes[ volumeGuid ] )

end

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

local function createQueryStr( keepAllOriginals, keepAllProxies, volumeStatuses )

	check( 'boolean', 'boolean', 'table' )

	local selectAndFromStmt = 
		[[SELECT localStatusAndOzStatus.docId, needsOriginalDownload, hasLocalOriginal, needsProxyDownload, 
			hasLocalProxy, proxySize, isMaybeDenied, isImportant, isMarkedForOffline,
			importDate, captureDate, waitingForProxyOnOz, needsDevelopXmpDownload, 
			ozHasNonThumbnailRendition, isVersionDocument
		FROM localStatusAndOzStatus]]

	local function expandSql( sql, volumeStatus, whereClause, maybeFilterIfOverIdleSpace )
	
		local overIdleSpace = maybeFilterIfOverIdleSpace and volumeStatus.overIdleSpaceThreshold
	
		-- Filter out the where clauses based on if the destination drive for the
		-- download is offline or out of space.
		
		if not volumeStatus.offline and not volumeStatus.outOfSpace and not overIdleSpace then
	
			if not sql then
				sql = string.format( "%s WHERE %s\n", selectAndFromStmt, whereClause )
			else
				sql = string.format( "%s UNION\n%s WHERE %s\n", sql, selectAndFromStmt, whereClause )
			end
	
		end
		
		return sql
	
	end
	
	local function expandSqlForCustomQuery( sql, volumeStatus, sqliteQuery )
		
		if not volumeStatus.offline and not volumeStatus.outOfSpace then
	
			if not sql then
				sql = string.format( "%s\n", sqliteQuery )
			else
				sql = string.format( "%s UNION\n%s\n", sql, sqliteQuery )
			end
	
		end
		
		return sql
	
	end
	
	local function expandSqlForIdleSpace( sql, volumeStatus, whereClause )
	
		return expandSql( sql, volumeStatus, whereClause, true )
	
	end
	
	local originalVolumeStatus = volumeStatuses[ 'original' ]
	local proxyVolumeStatus = volumeStatuses[ 'proxy' ]
	local xmpVolumeStatus = volumeStatuses[ 'xmp_develop' ]
	
	local queryStr
	
	if keepAllOriginals then
	
		queryStr = expandSql( queryStr, originalVolumeStatus, 'needsOriginalDownload == 1' )
			-- All photos that require originals.

		queryStr = expandSql( queryStr, proxyVolumeStatus, 'needsProxyDownload == 1 AND waitingForProxyOnOz == 0 AND hasLocalOriginal == 0 AND needsOriginalDownload == 0' )
			-- Photos that require the originals but have no originals, and so they require the proxies.

	else
	
		queryStr = expandSql( queryStr, originalVolumeStatus, 'needsOriginalDownload == 1 AND isMarkedForOffline == 1' )
			-- Photos that require originals because the photo is marked for offline use.
	
		queryStr = expandSqlForIdleSpace( queryStr, originalVolumeStatus, 'hasLocalProxy == 0 AND needsProxyDownload == 0 AND needsOriginalDownload == 1' )
			-- The rest of the photos that have no proxies but have originals, when space permits.	
	
	end

	if keepAllProxies then
	
		queryStr = expandSql( queryStr, proxyVolumeStatus, 'needsProxyDownload == 1 AND waitingForProxyOnOz == 0' )
			-- All photos that require proxies.

		queryStr = expandSql( queryStr, originalVolumeStatus, 'needsOriginalDownload == 1 AND hasLocalProxy == 0 AND needsProxyDownload == 0' )
			-- Photos that require the proxies but have no proxies, and so they require the originals.
		
	else
	
		queryStr = expandSql( queryStr, proxyVolumeStatus, 'needsProxyDownload == 1 AND waitingForProxyOnOz == 0 AND isMarkedForOffline == 1' )
			-- Photos that require proxies because the photo is marked for offline use.
		
		queryStr = expandSqlForIdleSpace( queryStr, proxyVolumeStatus, 'needsProxyDownload == 1 AND waitingForProxyOnOz == 0' )
			-- The rest of the proxies, when space permits.
	
	end

	if keepAllProxies or keepAllOriginals then
	
		queryStr = expandSql( queryStr, xmpVolumeStatus, 'needsDevelopXmpDownload == 1' )
			-- All photos/versions that have external xmp develop settings.
	
	else
	
		queryStr = expandSqlForIdleSpace( queryStr, xmpVolumeStatus, 'needsDevelopXmpDownload == 1 AND isVersionDocument == 0' )
			-- Photos (not versions) external xmp develop settings, as space permits.
			
		local versionsQuery = string.format( 
				[[%s 
					INNER JOIN versions ON versions.localId == localStatusAndOzStatus.docId
					WHERE needsDevelopXmpDownload == 1
					AND versions.assetId IN 
						( SELECT assetID FROM assets WHERE assets.isMarkedForOffline = 1 )]], selectAndFromStmt )
		
		queryStr = expandSqlForCustomQuery( queryStr, xmpVolumeStatus, versionsQuery )
			-- All versions whose associated asset is marked for offline use.
	
	end
	
	queryStr = expandSql( queryStr, proxyVolumeStatus, 'ozHasNonThumbnailRendition == 0 AND needsProxyDownload == 1 AND hasLocalOriginal == 0 AND waitingForProxyOnOz == 0' )
		-- Photos that require a proxy because there's no sufficient rendition on Oz.
	
	queryStr = expandSql( queryStr, originalVolumeStatus, 'ozHasNonThumbnailRendition == 0 AND hasLocalProxy == 0 AND needsOriginalDownload == 1' )
		-- Photos that require an original because there's no sufficient rendition on Oz
		
	return queryStr

end

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

local function buildDownloadQuery( self, volumeStatuses )

	local limitBuffer = 25
		-- SPAM [jandrew 05/24/2016]: In tests, increasing the limit buffer from
		-- 10 to 25 decreased the download time for 10,000 assets from about 
		-- ~215 seconds to ~180 seconds.

	local needsDownloadQueryStream = Observable.just( {} )

	local queryStr = createQueryStr( self._keepAllOriginals, self._keepAllProxies, volumeStatuses )

	if queryStr then
	
		local needsDownloadQueryStr = string.format( 
				[[%s
				ORDER BY isMaybeDenied ASC, isMarkedForOffline DESC, isVersionDocument ASC, 
						hasLocalOriginal ASC, ozHasNonThumbnailRendition ASC, isImportant DESC, 
						importDate DESC, captureDate DESC
				LIMIT %d]],
				queryStr,
				self._debugQueryLimit or maxScheduledIdleDownloads * limitBuffer )
		
		logger:infof( "needsDownloadQuery: %s", needsDownloadQueryStr )
		
		-- SPAM [jandrew 2017-02-24]: We might want to create a generalized wrapper for the
		-- following calm -> uncalmed logic, as it could be useful in a few other tasks.
		
		------------------------------------------------------------------------

		-- needsDownloadQueryStream is a stream that alternates between
		-- a calmed index store query and a fully open ( not-calmed ) index
		-- store query. When we have data to process, we use a calmed query
		-- that fires, at most, every 30 seconds. 
		
		-- When we run out of data to process, we immediately switch to a query that
		-- is not calmed. This way, we immediately receive new data to process once 
		-- it becomes available.
		
		-- Once there is data to process, we switch back to a calmed query to avoid
		-- unnecessary updates and to reduce the burden on the index store.
		
		-- This solution dramatically reduces the total number of index store queries 
		-- required by the idle downloader, while also preventing the idle downloader 
		-- from needing to make periodic calls to the index store to check for new data 
		-- when it runs out of work.
		
		local isQueryCalmed = false
		self._shouldCalmQueryStream:onNext( true )
		
		needsDownloadQueryStream = self._shouldCalmQueryStream
					
			:filter( function( shouldCalmQuery )
			
				if not shouldCalmQuery and not isQueryCalmed then
					-- no-op
					return false
				elseif shouldCalmQuery and isQueryCalmed then
					-- no-op
					return false
				else
					return true
				end
			
			end )
				
			:flatMapLatest( function( shouldCalmQuery )
			
				isQueryCalmed = shouldCalmQuery
			
				-- Fire off a query to fetch new assets to download. This query
				-- will refresh every 30 seconds help prevent working with
				-- stale download data.
				
				return self._catalog:indexStore():query {
					sql = needsDownloadQueryStr,
					calm = shouldCalmQuery and 30 or nil
				}
				
			end )
			
	------------------------------------------------------------------------
			
		needsDownloadQueryStream = maybePerformanceWrap( 'query', needsDownloadQueryStream )
	
	end
	
	return needsDownloadQueryStream

end

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

local function getVolumeGuidForBinaryType( self, binaryType )
	
	check( 'WFCatIdleDownloader', 'string' )
	
	return self._volumes[ binaryType ]
	
end

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

local function initVolumes( self )

	local volumes = self._volumes

	if not volumes then
	
		volumes = {}
	
		volumes[ 'original' ] = self._originalsVolumeGuid
		
		volumes[ 'proxy' ] = self._proxiesVolumeGuid
		
		volumes[ 'xmp_develop' ] = self._catalogVolumeGuid
	
	end
	
	return volumes

end

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

-- Returns true if all the volumes are offline or out of space.

local function allVolumesOfflineOrOutOfSpace( volumeStatuses )

	check( 'table' )

	for _, volumeStatus in pairs( volumeStatuses ) do
	
		if not volumeStatus.offline and not volumeStatus.outofSpace then
			return false
		end
	
	end
	
	return true
	
end

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

local function maybeCancelSomeInFlightDownloads( self, volumeStatuses )

	check( 'WFCatIdleDownloader', 'table' )

	for binaryType, volumeStatus in pairs( volumeStatuses ) do

		if volumeStatus.offline or volumeStatus.outOfSpace then
		
			-- If the volume is offline or out of space, cancel all in-flight downloads
			-- associated with the binary type.
			
			abandonAllDownloadsOfType( self, binaryType )
			
		elseif volumeStatus.overIdleSpaceThreshold and binaryType ~= 'xmp_develop' then
		
			-- If the volume is over the idle space threshold, cancel all in-flight
			-- downloads that are not required. xmp_develop settings are downloaded
			-- regardless of idle space.
			
			abandonNonRequiredDownloadsOfType( self, binaryType )
			
		end
	
	end
	
end

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

local function toBool( zeroOrOne )

	return zeroOrOne > 0

end

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

local function downloadSomeBinaries( self )

	return self._statusOfVolumesStream
	
		:distinctUntilChanged( nil, distinctTableCompare )
			-- When the status changes for the volumes, we need to refresh our query.
			
		:flatMapLatest( function( volumeStatuses )
		
			maybeCancelSomeInFlightDownloads( self, volumeStatuses )
		
			if allVolumesOfflineOrOutOfSpace( volumeStatuses ) then
			
				-- All volumes are offline, or outOfSpace, so we can't download anything.
				
				return Observable.just( {} )
				
			end
			
			local needsDownloadQueryStream = buildDownloadQuery( self, volumeStatuses )
		
			-- The download code within the pixel provider is responsible for annotating
			-- the doc store with information about the downloaded binaries. These
			-- annotations are what triggers a refresh of the needsDownloadQuery. There's
			-- a slight race condition where the query might be refreshed before we've 
			-- determined on this end that the download has finished. When this happens,
			-- the below rx chain will do no work since we still think we have the maximum
			-- number of downloads in flight. This can potentialy cause sync to freeze. In
			-- order to remedy this, we use combineLatest to cause the rx chain to fire on
			-- query refresh, as well as when a download finishes.
		
			return Rx.Observable.combineLatest(
			
				function( docsThatNeedBinaries )
					return docsThatNeedBinaries
				end,
				needsDownloadQueryStream,
				self._downloadFinished,
				Rx.Observable.timer( 
					0, 
					triggerDownloadAttemptEveryNSeconds, 
					WFRxScheduler.sharedScheduler() 
				)
			)
			
		end ) 

		:flatMapLatest( function( docsThatNeedBinaries )

			return self._continuousDownloadTask:throwIfNotRunning( docsThatNeedBinaries )
				-- Need to throw as this function doesn't ever complete.

		end )

		:tap( function( docsThatNeedBinaries )

			logger:debugf( "downloadSomeBinaries raw query result = %s", docsThatNeedBinaries )

			local outOfWork = true

			for _, row in ipairs( docsThatNeedBinaries ) do

				if not self._continuousDownloadTask:isRunning() then break end

				local localId = row[ 1 ]
				local needsOriginalDownload = toBool( row[ 2 ] )
				local hasLocalOriginal = toBool( row[ 3 ] )
				local needsProxyDownload = toBool( row[ 4 ] )
				local hasLocalProxy = toBool( row[ 5 ] )
				local binarySize = row[ 6 ]
				local isMaybeDenied = toBool( row[ 7 ] )
				local isMarkedForOffline = toBool( row[ 9 ] )
				local waitingForProxyOnOz = row[ 12 ] ~= 0
				local needsDevelopXmpDownload = toBool( row[ 13 ] )
				local ozHasNonThumbnailRendition = toBool( row[ 14 ] )
				
				if isMaybeDenied then
					isMaybeDenied = self._failedDownloadDeniedListHelper:maybeResetDeniedListStatus( localId )
				end

				-- If the volume is out of space or offline, we can't download the asset.
				-- The query should automatically filter out the assets that can't be
				-- downloaded due to these conditions, but there's a slight delay
				-- between when the query is updated and when the volume status changes.
				
				local volumeGuid = getVolumeGuidForBinaryType( self, 'original' )
				local originalVolumeOfflineOrFull = false
				
				if not volumeGuid or isVolumeFullOrOffline( self, volumeGuid ) then
					originalVolumeOfflineOrFull = true
				end
				
				volumeGuid = getVolumeGuidForBinaryType( self, 'proxy' )
				local proxyVolumeOfflineOrFull = false
				
				if not volumeGuid or isVolumeFullOrOffline( self, volumeGuid ) then
					proxyVolumeOfflineOrFull = true
				end
				
				local maybeBinaryType, isDownloadRequired
				
				if not isMaybeDenied then 
				
					maybeBinaryType, isDownloadRequired = WFCatSyncStatus.whatToDownload(
							self._keepAllOriginals,
							self._keepAllProxies,
							needsOriginalDownload, 
							hasLocalOriginal, 
							needsProxyDownload, 
							hasLocalProxy, 
							needsDevelopXmpDownload,
							isMarkedForOffline,
							originalVolumeOfflineOrFull,
							proxyVolumeOfflineOrFull,
							ozHasNonThumbnailRendition,
							waitingForProxyOnOz )
							
					maybeRemoveDownloadRequiredStatus( self, localId, maybeBinaryType, isDownloadRequired )
						-- If an existing binary download is is progress and isDownloadRequired is removed,
						-- we want to remove the required status from the download so that the global
						-- sync status icon no longer shows the downloads as sync activity, since non-required
						-- idle downloads should be invisible to the user.
						
				end
				
				if maybeBinaryType == 'original' then
					binarySize = 'master'
				elseif maybeBinaryType == 'xmp_develop' then
					binarySize = 'xmp_develop'
				end
				
				volumeGuid = maybeBinaryType and getVolumeGuidForBinaryType( self, maybeBinaryType )
				
				if not volumeGuid then
				
					maybeBinaryType = nil
					
				end
				
				-- If we think we're over the space limit, don't download anything.
				
				if maybeBinaryType and ( self._estimatedRemainingSpace[ maybeBinaryType ] or 0 ) < 0 then
				
					maybeBinaryType = nil
				
				end
				
				-- If the binary isn't required, don't download it if we're over our idle space target.
				-- We always download xmp_develop settings, tho.
				
				if maybeBinaryType ~= 'xmp_develop' then
				
					if self._statusOfVolumes.volumesOverIdleSpaceThreshold[ volumeGuid ] and not isDownloadRequired then
					
						maybeBinaryType = nil
						
					end
					
				end
				
				-- Figure out of the binary should be downloaded.
				
				local shouldDownload = false
				
				if maybeBinaryType and not wasRecentlyDownloaded( self, localId, maybeBinaryType ) then
				
					shouldDownload = true
					outOfWork = false
				
				end
				
				-- If the binary should be downloaded and there's an open slot, download it!
				
				if self._downloadsInFlightCount < maxScheduledIdleDownloads and shouldDownload then
					
					self._continuousDownloadTask:taskDidWork()

					self._downloadsInFlightCount = self._downloadsInFlightCount + 1
					
					markAsRecentlyDownloaded( self, localId, maybeBinaryType )
					
					logger:infof( "Now downloading binary for local Id: %s", localId )
			
					cacheBinaryInfo( self, localId, maybeBinaryType, binarySize, 
							isMarkedForOffline, hasLocalOriginal, isDownloadRequired )
					
					self._downloadFunc( localId, maybeBinaryType, binarySize, isMarkedForOffline, hasLocalOriginal )
					
					if self._downloadsInFlightCount == maxScheduledIdleDownloads then break end

				end
				
			end

			if outOfWork then
				self._shouldCalmQueryStream:onNext( false )
			else
				self._shouldCalmQueryStream:onNext( true )
			end

			if self._downloadsInFlightCount == 0 then
				
				self._continuousDownloadTask:taskDidWork( false )
					-- Since we don't typically send onCompleted events we need to
					-- indicate when we don't do work.
			
			end

		end )

		:catch( function( errorMessage )

			return self._continuousDownloadTask:throwIfNotStopped( errorMessage )

		end )

end


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

--[[!

	\class WFCatIdleDownloader
	
	\brief This namespace can be used to identify and start the low-priority 
		downloads of binary assets that are not present on disk.
		
	The Idle Downloader operates differently based on the download strategy.
	
	----------------------------------------------------------------------------
	
	Default
	
	LrCC will fill up to 50% of the target space by idly downloading smart previews. 
	We first download the active SPs, then the inactive SPs if there's room. 
	The rest of the free-for-all window can be filled with by importing, 
	or downloading negatives as the result of some user action.
	
	----------------------------------------------------------------------------
	
	Keep All Originals
	
	The idle downloader will download all originals.
	
	After the originals are downloaded, some portion of the remaining space is reserved for
	Smart Previews. Half of this space will be filled by idle Smart Preview downloads.
	The remainder can be filled by on-demand Smart Preview downloads.
	
	----------------------------------------------------------------------------
	
	Keep All Smart Previews
	
	The idle downloader will download all Smart Previews.
	
	After the smart previews are downloaded, 50% of the remaining space is reserved for Originals.
	This space can only be filled by on-demand original downloads and imports.
	
]]--

-- @mark WFCatIdleDownloader namespace

local WFCatIdleDownloader = {

	_catalog = false,
		-- (WFCatalog)
	
	_originalsVolumeGuid = false,
		-- (string) The GUID of the volume where originals are downloaded.
	
	_proxiesVolumeGuid= false,
		-- (string) The GUID of the volume where proxies are downlaoded.
		
	_catalogVolumeGuid = false,
		-- (string) The GUID of the volume where the Nimbus catalog lives; where renditions and XMP are downloaded.
		
	_downloadFunc = false,
		-- (function) A user-specified function used to perform the actual binary downloading.
		
	_batchCancelDownloadsFunc = false,
		-- (function) A user-specified function used to batch cancel in-flight binary downloads.

	_downloadStream = false,
		-- (Observable< table >) A stream that fires as binaries are downloaded.

	_keepAllOriginals = false,
		-- (boolean) If true, download all originals.
		
	_keepAllProxies = false,
		-- (boolean) If true, download all proxies.
		
	_downloadStreamSubscription = false,
		-- (Rx Subscription) A subscription to the download stream.

	_downloadsInFlightCount = false,
		-- (number) The current number of downloads in progress.
		
	_recentlyDownloadedBinaries = false,
		-- (table) Lookup table of Local IDs of the binary downloads that are in progress or recently in progress.
		-- key: Comination of local id and binary type (<local id>::<binary type>)
		-- value: true
	
	_binaryInfoCache = false,
		-- (table) Lookup table to associate local Id with binary type and size.
		-- key: Local ID
		-- value: { binary type, binary size }
		
	_estimatedRemainingSpace = false,
		-- (table) A table containing how much space we think we have remaining for each binary type.
		-- key: binary type ( original, proxy, xmp )
		-- value: estimated amount of space avilable for the given binary type.
			
	_isDownloadingRequiredBinaryStream = false,
		-- (Rx ReplaySubject) A stream that fires a boolean indicating whether or not the downloader
		-- is actively downloading a required binary.
		
	_requiredDownloadsInFlight = false,
		-- (table) The required binary downloads that are currently in flight.
		
	_downloadFinished = false,
		-- (Rx ReplaySubject) A local stream that fires when a download completes.
		
	_continuousDownloadTask = false,
		-- (WFRxContinuousTask) A continuous task to handle continuous downloads.

	_continuousWaitingForProxiesTask = false,
		-- (WFRxContinuousTask) A continous task which monitors assets that are waiting for proxies
		-- to be generated on Oz.

	_waitingForProxyTimersSubscriptions = false,
		-- (table) A table containing timer subscriptions for assets that are waiting for proxies on Oz.
		-- key: assetId
		-- value: Rx.Disposable

	_failedDownloadDeniedListHelper = false,
		-- (WFCatDeniedListHelper) A helper classed used to temporarily deny/block failed downloads.

	_volumes = false,
		-- (table) A table of all volumes used to store Nimbus assets.
		-- key: binary type stored on the volume ( original, proxy, xmp )
		-- value: GUID of the volume
		
	_statusOfVolumes = false,
		-- (table) A table of volume statuses including connection status, and disk
		-- space status.
	
	_statusOfVolumesStream = false,
		-- (Rx ReplaySubject) A local stream that fires when a volumes' status changes.

	_shouldCalmQueryStream = false,
		-- (Rx ReplaySubject) A local stream that fires when we need to calm or uncalm the
		-- index store query that fetches what to dwonload.

	_binariesDownloadedByTypeStream =  false,
		-- (Rx BehaviorSubject) Fires whenever _binariesDownloadedByType is updated. 
	
	_binariesDownloadedByType = false,
		-- (table) Table of the number of binaries downloaded, organized by type.

	_debugQueryLimit = false,
		-- (number) Only used by unit tests to modifying the number of assets returned by the
		-- index store query used to figure out what to download.

	class = {
		generateIndexSchema = generateIndexSchema,
		addSearchToSearchSchema = addSearchToSearchSchema,
		addWaitingForProxyAnnotation = addWaitingForProxyAnnotation,
	},

	WFCloseSupport.addCloseSupportMethods,

}

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

local function monitorWaitingForProxyAssets( self )

	return self._catalog:indexStore():query {
			sql = [[SELECT docId, ozHasProxy, subtype, waitingForProxyOnOz
					FROM localStatusAndOzStatus
					WHERE waitingForProxyOnOz > 0 ]],
			calm = 5,
		}

		:flatMapLatest( function( rows )

			return self._continuousWaitingForProxiesTask:throwIfNotRunning( rows )
				-- Need to throw as this function doesn't ever complete.

		end )

		:map( function( rows )
		
			for _, row in ipairs( rows ) do
			
				if not self._continuousWaitingForProxiesTask:isRunning() then break end
			
				local assetId = row[ 1 ]
				local ozHasProxy = toBool( row[ 2 ] )
				local subtype = row[ 3 ]
				local waitingForProxyOnOz = row[ 4 ]
				
				local isVideo = subtype == 'video' or subtype == 'deleted_video'
				local waitTimeInSeconds = isVideo and 120 or 60
				
				-- If the asset has a proxy, or if we've already waited long enough,
				-- then clear the waiting-for-proxy status.
				
				if ozHasProxy or not dateIsWithinNSeconds( waitingForProxyOnOz, waitTimeInSeconds ) then
				
					-- If there's an existing wait timer, delete it.
					
					local maybeWaitTimerSubscription = self._waitingForProxyTimersSubscriptions[ assetId ]
					self._waitingForProxyTimersSubscriptions[ assetId ] = nil
					
					if maybeWaitTimerSubscription then
						maybeWaitTimerSubscription:dispose()
					end
					
					-- Clear the waiting-for-proxy status
					
					clearWaitingForProxyStatus( self._catalog:docStore(), assetId )
					
				else
				
					-- If there's already an existing wait timer, do nothing.
					
					local maybeWaitTimerSubscription = self._waitingForProxyTimersSubscriptions[ assetId ]
				
					if not maybeWaitTimerSubscription then
					
						-- Create a new wait timer that fires when the waiting-for-proxy status
						-- should be cleared.
						
						local adjustedWaitTime = waitTimeInSeconds
						
						local currentTime = KSDate.timeNow()
						
						local ok, waitinForProxyTimestampInSeconds = pcall( KSDate.iso8601ToTimeIntervalSinceReference, waitingForProxyOnOz )
						
						if ok then
						
							adjustedWaitTime = math.max( waitTimeInSeconds, 
									( waitinForProxyTimestampInSeconds + waitTimeInSeconds ) - currentTime )
									
						end
					
						local waitTimerSubscription = WFRxLowFidelityTimer.timer( adjustedWaitTime )
							:take( 1 )
							:subscribe( function()
							
								clearWaitingForProxyStatus( self._catalog:docStore(), assetId )
								self._waitingForProxyTimersSubscriptions[ assetId ] = nil
								
							end )
							
						self._waitingForProxyTimersSubscriptions[ assetId ] = waitTimerSubscription
					
					end
				
				end

			end
			
			return true
			
		end )

		:catch( function( errorMessage )

			return self._continuousWaitingForProxiesTask:throwIfNotStopped( errorMessage )

		end )

end

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

--[[!

	\brief Creates a new idle-binary downloader instance.

	\param catalog (WFCatalog) A valid WFCatalog object.
	\param downloadFunc (function) A function that does the actual downloading
		of the binary from Oz.
	\param downloadStream (Rx.Observable) A stream that fires when a binary
		has been successfully downloaded via WFCatIdleDownloader.
	\param batchCancelDownloadsFunc (function) A function used to batch cancel in-flight
		downloads of a certain type.
	\param keepAllOriginals (boolean) Whether or not we should download all originals.
	\param keepAllProxies (boolean) Whether or not we should download all proxies.
	\param optArgs (?table) A table of optional arguments only used by unit tests.
	
	\return WFCatIdleDownloader object
	
--]]

function WFCatIdleDownloader:__init( catalog, downloadFunc, downloadStream, 
		batchCancelDownloadsFunc, keepAllOriginals, keepAllProxies, optArgs )

	check( 'WFCatIdleDownloader', 'WFCatalog', 'function', 'table', 'function', 'boolean', 'boolean', '?table' )

	logger:infof( "Constructing idle binary downloader for catalog %s", tostring( catalog ) )

	self:setOwned {
	
		_failedDownloadDeniedListHelper = WFCatDeniedListHelper( 
				catalog:docStore(), 
				'WFCatIdleDownloader',
				numberOfTriesBeforeBlockingBinary,
				failedDownloadBlackoutWindowInSeconds,
				true ),
	
		_catalog = catalog:relayObject()
		
	}

	optArgs = optArgs or {}

	self._originalsVolumeGuid = optArgs.originalsVolumeGuid or catalog:fileSpec():originalsPathInfo().volumeGuid
	self._proxiesVolumeGuid = optArgs.proxiesVolumeGuid or catalog:fileSpec():proxiesPathInfo().volumeGuid
	self._catalogVolumeGuid = optArgs.catalogVolumeGuid or catalog:fileSpec():catalogPathInfo().volumeGuid
	
	self._debugQueryLimit = optArgs.debugQueryLimit
	
	downloadBlackoutWindowInSeconds = optArgs.downloadBlackoutWindowInSeconds or downloadBlackoutWindowInSeconds
	
	self._volumes = initVolumes( self )

	self._statusOfVolumes = {}

	self._statusOfVolumesStream = ReplaySubject( 1 )

	self._downloadFunc = downloadFunc

	self._batchCancelDownloadsFunc = batchCancelDownloadsFunc

	self._downloadStream = downloadStream

	self._keepAllOriginals = keepAllOriginals

	self._keepAllProxies = keepAllProxies

	self._downloadFinished = ReplaySubject( 1 )
	self._downloadFinished:onNext( true )

	self._shouldCalmQueryStream = ReplaySubject( 1 )
	self._shouldCalmQueryStream:onNext( true )
	
	self._isDownloadingRequiredBinaryStream = ReplaySubject( 1 )
	self._isDownloadingRequiredBinaryStream:onNext( false )
	
	self._binariesDownloadedByTypeStream = BehaviorSubject( {} )
	self._binariesDownloadedByType = {}
	
	self._requiredDownloadsInFlight = {}
	
	self._downloadsInFlightCount = 0

	self._recentlyDownloadedBinaries = {}

	self._estimatedRemainingSpace = {}

	self._binaryInfoCache = {}

	setupDownloadListener( self )

	-- Download Task
	
	local downloadTaskDelays = {
		progressiveBackoff = true,
		errorDelay = 5,
		maxDelay = 60,
	}
	
	local continuousDownloadTask = function() return downloadSomeBinaries( self ) end
	
	self._continuousDownloadTask = WFRxContinuousTask( "WFCatIdleDownloader", continuousDownloadTask, downloadTaskDelays )

	-- Monitor Waiting For Proxy Assets Task
	
	self._waitingForProxyTimersSubscriptions = {}
	
	local waitForProxiesTaskDelays = {
		progressiveBackoff = false,
		errorDelay = 10,
		maxDelay = 10,
	}
	
	local continuousWaitForProxiesTask = function() return monitorWaitingForProxyAssets( self ) end
	
	self._continuousWaitingForProxiesTask = WFRxContinuousTask( "MonitorWaitingForProxyAssets", continuousWaitForProxiesTask, waitForProxiesTaskDelays )
	self._continuousWaitingForProxiesTask:startContinuousTask()
	
	WFDiagnosticLog.addLogProvider( self )
	
end

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

--[[!

	\brief Close and stop the idle-binary downloader instance.

	\return (void)
	
--]]

function WFCatIdleDownloader:close()

	check( 'WFCatIdleDownloader' )
	
	local downloadStreamSubscription = self._downloadStreamSubscription
	self._downloadStreamSubscription = nil
	
	if downloadStreamSubscription then
		downloadStreamSubscription:dispose()
	end

	local continuousDownloadTask = self._continuousDownloadTask
	self._continuousDownloadTask = nil
	
	if continuousDownloadTask then
		continuousDownloadTask:close()
	end

	for key, subscription in pairs( self._waitingForProxyTimersSubscriptions ) do
	
		subscription:dispose()
		self._waitingForProxyTimersSubscriptions[ key ] = nil
		
	end

	local continuousWaitingForProxiesTask = self._continuousWaitingForProxiesTask
	self._continuousWaitingForProxiesTask = nil
	
	if continuousWaitingForProxiesTask then
		continuousWaitingForProxiesTask:close()
	end

	local shouldCalmQueryStream = self._shouldCalmQueryStream
	self._shouldCalmQueryStream = nil
	
	if shouldCalmQueryStream then
		shouldCalmQueryStream:dispose()
	end
	
	local isDownloadingRequiredBinaryStream = self._isDownloadingRequiredBinaryStream
	self._isDownloadingRequiredBinaryStream = nil
	
	if isDownloadingRequiredBinaryStream then
		isDownloadingRequiredBinaryStream:dispose()
	end

	local binariesDownloadedByTypeStream = self._binariesDownloadedByTypeStream
	self._binariesDownloadedByTypeStream = nil
	
	if binariesDownloadedByTypeStream then
		binariesDownloadedByTypeStream:dispose()
	end

end

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

--[[!

	\brief Stop the continuous idle downloads of binaries.

	\return (void)
	
--]]

function WFCatIdleDownloader:stop()

	check( 'WFCatIdleDownloader' )

	-- Stop the downloader
	
	if self._continuousDownloadTask then 
		self._continuousDownloadTask:stopContinuousTask()
	end

	-- Cancel all in-flight downloads.

	abandonAllDownloadsOfType( self, 'original' )
	abandonAllDownloadsOfType( self, 'proxy' )
	abandonAllDownloadsOfType( self, 'xmp_develop' )

end

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

--[[!

	\brief Start the continuous idle downloads of binaries.

	\return (void)
	
--]]

function WFCatIdleDownloader:start()

	check( 'WFCatIdleDownloader' )

	if self._continuousDownloadTask then
		self._continuousDownloadTask:startContinuousTask()
	end

end

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

--[[!

	\brief Report the status of the Nimbus Volumes
	
	\return (void)
	
--]]

function WFCatIdleDownloader:reportStatusOfVolumes( statusOfVolumes )

	check( 'WFCatIdleDownloader', 'table' )
	self._statusOfVolumes = statusOfVolumes
	reportVolumeStatusChange( self )
	
end

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

--[[!

	\brief Reports whether the download task is downloading a required binary.
		In this context, 'required' means that the binary is either explicitly
		marked for offline use OR the it's a binary whose type (original|proxy)
		we're keeping all of locally.

	\return (Observable) true if downloading required binaries, otherwise false.
	
--]]

function WFCatIdleDownloader:isDownloadingRequiredBinary()

	check( 'WFCatIdleDownloader' )
	
	return Observable.combineLatest( 
			self._continuousDownloadTask:isTaskDoingWork(), self._isDownloadingRequiredBinaryStream )
		:map( function( isDoingWork, isDownloadingRequiredBinary )
		
			return isDoingWork and isDownloadingRequiredBinary
		
		end )

end

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

--[[!

	\brief Explicitly clear the recently downloaded status used to prevent the system from
	redownloading the same assets repeatedly.

	\param localId (number) The local id of the asset to clear.
	\param binaryType (string) The binary type of the asset to clear.

	\return (void)
	
--]]

function WFCatIdleDownloader:clearRecentlyDownloadedStatus( localId, binaryType )

	check( 'WFCatIdleDownloader', 'number', 'string' )
	clearRecentlyDownloaded( self, localId, binaryType )
	
	self._downloadFinished:onNext( true )
		-- Force the download chain to execute.
		
end

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

local function createDiagnosticSummaryStr( binariesDownloadedByType )

	local numberOfOriginals
			= derefOpt( binariesDownloadedByType, 'original' ) or 0
	local numberOfProxies
			= derefOpt( binariesDownloadedByType, 'proxy' ) or 0
	local numberOfXmpDevelopSettings
			= derefOpt( binariesDownloadedByType, 'xmp_develop' ) or 0

	return string.format( "Downloaded %d originals, %d proxies, %d xmp develop settings from Oz.", 
			numberOfOriginals, numberOfProxies, numberOfXmpDevelopSettings )
			
end

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

--[[!
	\brief This API that can be used gather real-time diagnostic data from the
		idle downloader.

	\param optCalm (optional number) The maximum frequency at which the stream will
		fire with updates. Defaults to 1 second. 
	
	\return Observable< table > An Observable whose value tracks a table of
		sync diagnostic data. An example data event may look like this:
		
		Please note that the idle downloader uses local IDs, and not full document IDs.
		
		Also note that downloads will disapear from the recentAndScheduledDownloads section
		after 30 seconds.
		
	{
		downloadsByType = {
			original = 4,
			proxy = 58,
		},
		isCurrentlySyncing = true,
		numberOfRequiredBinariesRemainingToDownload = 1,
		numberOfScheduledDownloads = 4,
		recentAndScheduledDownloads = {
			"2008::original",
			"2791::original",
			"2006::proxy",
			"5650::proxy",
			"2893::proxy",
			"5608::proxy",
			"2006::original",
			"4834::proxy",
		},
		scheduledDownloads = {
			[2006] = {
				binarySize = "2560",
				binaryType = "proxy",
				downloadRequired = true,
				hasLocalOriginal = true,
				isMarkedForOffline = true,
			},
			[2007] = {
				binarySize = "2560",
				binaryType = "proxy",
				downloadRequired = true,
				hasLocalOriginal = true,
				isMarkedForOffline = true,
			},
			[2008] = {
				binarySize = "master",
				binaryType = "original",
				downloadRequired = true,
				hasLocalOriginal = false,
				isMarkedForOffline = true,
			},
			[2791] = {
				binarySize = "master",
				binaryType = "original",
				downloadRequired = true,
				hasLocalOriginal = false,
				isMarkedForOffline = true,
			},
		},
		summaryStr = "Downloaded 4 originals, 58 proxies, 0 xmp develop settings from Oz.",
	}
		
--]]

function WFCatIdleDownloader:syncDiagnostics( optCalm )

	check( 'WFCatIdleDownloader', '?number' )

	local syncStatusColumnName = WFCatSyncStatus.getSyncStatusColumnName()

	local numberOfRequiredBinariesRemainingToDownloadStream = self._catalog:indexStore():query( {
				sql = string.format( [[SELECT COUNT( * ) FROM asset_sync_status_ui WHERE %s IN ( 3, 4, 5, 6, 25 )]], syncStatusColumnName ),
				calm = optCalm or 1,
				singleColumnQuery = true,
			} )
			
			:map( function( rows )
			
				return derefOpt( rows, 1 ) or 0
			
			end )

	return Observable.combineLatest(
			self._binariesDownloadedByTypeStream,
			self._continuousDownloadTask:isTaskDoingWork(),
			numberOfRequiredBinariesRemainingToDownloadStream )

		:calm( optCalm or 1, WFRxScheduler.sharedScheduler() )

		:map( function( binariesDownloadedByType, isCurrentlySyncing, numberOfRequiredBinariesRemainingToDownload )
		
			local diagnostics = {
				summaryStr = createDiagnosticSummaryStr( binariesDownloadedByType ),
				isCurrentlySyncing = isCurrentlySyncing,
				numberOfScheduledDownloads = self._downloadsInFlightCount,
				downloadsByType = binariesDownloadedByType,
				scheduledDownloads = self._binaryInfoCache,
				recentAndScheduledDownloads = KSTableUtils.extractKeys( self._recentlyDownloadedBinaries ),
				numberOfRequiredBinariesRemainingToDownload = numberOfRequiredBinariesRemainingToDownload
			}
			
			logger:infof( "Idle Downloader syncDiagnostics update: %s", diagnostics )
			
			return diagnostics
				
		end )

end

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

function WFCatIdleDownloader:sendDiagnosticLogData()

	check( 'WFCatIdleDownloader' )

	return {
		
		name = 'Idle Binary Downloads',
		id = 'WFCatIdleDownloader',
		weight = 500,
		parentSection = WFDiagnosticLog.Sections.SYNC,

		body = {
			diagnostics = self:syncDiagnostics():take( 1 )
		}

	}

end

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

return Rx.Class 'WFCatIdleDownloader'( WFCatIdleDownloader )
