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

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

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

--[[
	AdobePatentID="P7167-US"
]]

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

	-- WF.oz_client.managed_catalog
local WFCatDelete = require 'WFCatDelete'
local WFCatPhotoImportance = require 'WFCatPhotoImportance'
local NmbFileManagementSettings = require 'NmbFileManagementSettings'

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

	-- WF.core.utils
local KSDate = require 'KSDate'
local KSIdGenerator = require 'KSIdGenerator'
local KSPathUtils = require 'KSPathUtils'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFCloseSupport = require 'WFCloseSupport'
local WFConfiguration = require 'WFConfiguration'
local WFFileUtils = require 'WFFileUtils'
local WFRingBuffer = require 'WFRingBuffer'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'
 
--------------------------------------------------------------------------------

local check = assert( WFArguments.check )

local child = assert( KSPathUtils.child )

local runUntilDone = WFRxRunUntilDone

local KSTableUtils_derefOpt = KSTableUtils.derefOpt

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

local diskSpaceCheckPeriod = 60
	-- Even if there are no changes within the app, we run the disk space check 
	-- periodically to check if anything changed outside the app. This value determines
	-- how frequently we run that check.

local reactivePurgeCalmInterval = 120
	-- At most, run reactive purge this often.

local volumeStatusCheckPeriod = 5
	-- Every N seconds, check the connection status of the volume.

local idleSpaceThreshold = 0.5
	-- The percentage of disk space that may be used by the idle downloader.
	
local reclaimTargetSpaceBuffer = 0.10
	-- The percentage of of the target range that is attempted to be reclaimed by reactive purge.

local maxDebugLogSize = 100
	-- The maximum size of the debug log that is included in the diagnostic log.
	
--------------------------------------------------------------------------------

local reactivePurgeReasons = {

	NO_DISK_SPACE 					= 1,
	TARGET_RANGE 					= 2,
	NO_DISK_SPACE_ALL_OR_NOTHING 	= 3,
	PROACTIVE_PROXIES				= 4,
	PROACTIVE_ORIGINALS				= 5,
	
}

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

local function logInfof( self, str, ... )

	WFRingBuffer.addItem( self._debugLogRingBuffer, string.format( str, ... ) )

	self._logger:infof( str, ... )
		
end

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

local function logWarnf( self, str, ... )

	WFRingBuffer.addItem( self._debugLogRingBuffer, string.format( str, ... ) )

	self._logger:warnf( str, ... )
		
end

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

local function isAllOrNothing( reason )

	return reason == reactivePurgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING
		
end

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

local function isProactivePurge( reason )

	return reason == reactivePurgeReasons.PROACTIVE_PROXIES or reason == reactivePurgeReasons.PROACTIVE_ORIGINALS
		
end

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

local function bytesToGb( bytes )
	return bytes / 1024 / 1024 / 1024
end

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

local function buildAbsoluteNimbusPath( self )

	local path = WFFileUtils.createFullPathFromRelativePathAndGuid( self._nimbusPath, self._volumeGuid )

	if not path then
		logInfof( self, "Failed to build absolute Nimbus path from %s and volume GUID %s", self._nimbusPath, self._volumeGuid )
	end
	
	return path
	
end

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

local function maybeClearRecentlyDownloadedStatus( self, localId, binaryType )

	if self._idleDownloader then
		self._idleDownloader:clearRecentlyDownloadedStatus( localId, binaryType )
	end
	
end

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

local function deleteProxies( self, localIds )

	if #localIds > 0 then
		logInfof( self, "Now purging %d proxies.", #localIds )
	end
	
	return WFCatDelete.deleteLocalProxiesByDocIds( self._docStore, localIds, self._deletionBacklog )
	
		:tap( function()
		
			for _, localId in ipairs( localIds ) do
			
				maybeClearRecentlyDownloadedStatus( self, localId, 'proxy' )
					-- Allow the asset to be immediately redownloaded, if needed.
			
			end
		
		end )
		
end

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

local function deleteOriginals( self, localIds )

	if #localIds > 0 then
		logInfof( self, "Now purging %d originals.", #localIds )
	end
	
	return WFCatDelete.deleteLocalOriginalsByDocIds( self._docStore, localIds, self._deletionBacklog )
		
		:tap( function()
		
			for _, localId in ipairs( localIds ) do
				
				maybeClearRecentlyDownloadedStatus( self, localId, 'original' )
					-- Allow the asset to be immediately redownloaded, if needed.
			
			end
		
		end )
		
end

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

local function isOriginalPurgeable( volumeGuidUsedByProxy, willEventuallyPurgeProxies )

	-- If we're only purging originals and not smart previews, then we don't want to purge the original 
	-- if there's no Smart Preview to replace it with or we may end up in a download/purge loop.

	if not volumeGuidUsedByProxy and not willEventuallyPurgeProxies then
		return false
	end

	return true

end

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

local function isProxyPurgeable( keepAllOriginals, ozHasOriginal )

	-- Treat the proxy as it is the original if we're puring proxies and oz
	-- doesn't have an original (especially for the case for assets from classic
	-- that won't upload an original as indicated by the willNotUpload flag in
	-- the import section).

	if not ozHasOriginal and keepAllOriginals then
		return false
	end

	return true

end

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

local function isProtectedFromPurge( self, localId, docId )

	if ( localId and self._protectedFromPurge[ localId ] ) or
		( docId and self._protectedFromPurge[ docId ] ) then
	
		logInfof( self, "%s is protected from purge.", localId )
		return true
		
	end

	return false

end

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

local function maybePurgeLeastRecentlyAccessed( self, whatToPurge, willEventuallyPurgeProxies, 
		spaceNeeded, leastRecentlyAccessed, getNextBatch, purgeList, purgeStream, maxSizePurgeable )

	check( 'NmbVolumeAgent', 'string', 'boolean', 'number', 'table', 'function', 'table', 'table', '?number' )

	maxSizePurgeable = maxSizePurgeable or math.huge

	if #leastRecentlyAccessed == 0 then
		logInfof( self, "Nothing to purge..." )
		purgeStream:onNext( purgeList, spaceNeeded )
		purgeStream:onCompleted()
		return
	end
	
	logInfof( self, "Processing %d assets to maybe purge.", #leastRecentlyAccessed )
	
	local purgeOriginals = false
	local purgeProxies = false

	if whatToPurge == 'original' then
		purgeOriginals = true
	elseif whatToPurge == 'proxy' then
		purgeProxies = true
	else
		purgeStream:onNext( purgeList, spaceNeeded )
		purgeStream:onCompleted()
		return
	end

	-- Fetch the doc values for the current batch of photos
	
	for _, assetInfo in ipairs( leastRecentlyAccessed ) do
	
		local localId = assetInfo[ 1 ]
		local docId = assetInfo[ 2 ]
		local proxyFileSize = assetInfo[ 3 ] or 0
		local originalFileSize = assetInfo[ 4 ] or 0
		local xmpSidecarSize = assetInfo[ 5 ] or 0
		local volumeGuidUsedByProxy = assetInfo[ 6 ]
		local ozHasOriginal = assetInfo[ 7 ] == 1
		
		if spaceNeeded <= 0 then
			break
		end

		local maybeSize
		
		if purgeOriginals then
			maybeSize = originalFileSize + xmpSidecarSize
		else
			maybeSize = proxyFileSize or 0
		end
		
		if maxSizePurgeable - maybeSize < 0 then

			-- maxSizePurgeable is the total amount of space that can be purged for the current binary type.
			-- For proxies, maxSizePurgeable is the total size of local proxies minus any space
			-- reserved by the idle downloader ( ~50% of the target disk space usage ).

			maxSizePurgeable = 0
			break

		end

		if not isProtectedFromPurge( self, localId, docId ) then

			local isOriginalMarkedForPurge = self._isOriginalMarkedForPurgeByLocalId[ localId ]
			local isProxyMarkedForPurge = self._isProxyMarkedForPurgeByLocalId[ localId ]
			
			if purgeOriginals and not isOriginalMarkedForPurge then

				if isOriginalPurgeable( volumeGuidUsedByProxy, willEventuallyPurgeProxies ) then

					self._isOriginalMarkedForPurgeByLocalId[ localId ] = true
						-- Mark the original. This is used to prevent purging the same
						-- asset more than once.

					table.insert( purgeList.originals, localId )
					
					spaceNeeded = spaceNeeded - maybeSize
					maxSizePurgeable = maxSizePurgeable - maybeSize

				end

			elseif purgeProxies and not isProxyMarkedForPurge then

				if isProxyPurgeable( self._keepAllOriginals, ozHasOriginal ) then

					self._isProxyMarkedForPurgeByLocalId[ localId ] = true
						-- Mark the proxy. This is used to prevent purging the same
						-- asset more than once.

					table.insert( purgeList.proxies, localId )
					spaceNeeded = spaceNeeded - maybeSize
					maxSizePurgeable = maxSizePurgeable - maybeSize

				end

			end
		
		end
	
	end
		
	if spaceNeeded > 0 and maxSizePurgeable > 0 then
	
		-- Get the next batch of assets within the current purgeWindow and continue purging.
		
		Observable.defer( function()
		
			return getNextBatch()

				:tap( function( newLeastRecentlyAccessed, newGetNextBatch )
			
					maybePurgeLeastRecentlyAccessed( self, whatToPurge, willEventuallyPurgeProxies, spaceNeeded, 
							newLeastRecentlyAccessed, newGetNextBatch, purgeList, purgeStream, maxSizePurgeable )
			
				end )
				
		end )
		
		:call( runUntilDone )
		
	else
	
		purgeStream:onNext( purgeList, spaceNeeded )
		purgeStream:onCompleted()
	
	end
		
end

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

-- Remove any purgeable assets from disk that fall into the current purgeWindow
-- until we reclaim enough disk space, or until we run out of assets to purge.

local function reactivelyPurgeSomeBinaries( self, whatToPurge, reason, willEventuallyPurgeProxies, 
		spaceNeeded, purgeWindow, purgeList, maxSizePurgeable )

	check( 'NmbVolumeAgent', 'string', 'number', 'boolean', 'number', 'table', 'table', '?number' )

	-- Fetch a batch of least-recently-accessed photos that fall into the current
	-- purge window.

	local batchSize = 10000
	
	-- Prevent purging the N ( _skipPurgeOfNMostRecentAssets ) most recently accessed assets, 
	-- no matter what. This is to prevent purging images that are currently being viewed in loupe.
	
	local leastRecentlyAccessedStream

	if whatToPurge == 'proxy' then
	
		local getLeastRecentlyAccessedFunc = self._fileAccessDb.getLeastRecentlyAccessedProxiesInWindow
	
		if isProactivePurge( reason ) then
			getLeastRecentlyAccessedFunc = self._fileAccessDb.getInactiveProxiesInWindow
		end
	
		leastRecentlyAccessedStream = getLeastRecentlyAccessedFunc( 
				self._fileAccessDb, self._volumeGuid, purgeWindow, batchSize, self._skipPurgeOfNMostRecentAssets )
	
	elseif whatToPurge == 'original' then
	
		local getLeastRecentlyAccessedFunc = self._fileAccessDb.getLeastRecentlyAccessedOriginalsInWindow
	
		if isProactivePurge( reason ) then
			getLeastRecentlyAccessedFunc = self._fileAccessDb.getInactiveOriginalsInWindow
		end
	
		leastRecentlyAccessedStream = getLeastRecentlyAccessedFunc( 
				self._fileAccessDb, self._volumeGuid, purgeWindow, batchSize, self._skipPurgeOfNMostRecentAssets )
	
	else
	
		local getLeastRecentlyAccessedFunc = self._fileAccessDb.getLeastRecentlyAccessedInWindow
	
		if isProactivePurge( reason ) then
			getLeastRecentlyAccessedFunc = self._fileAccessDb.getInactiveAssetsInWindow
		end
	
		leastRecentlyAccessedStream = getLeastRecentlyAccessedFunc( 
				self._fileAccessDb, self._volumeGuid, purgeWindow, batchSize, self._skipPurgeOfNMostRecentAssets )
	
	end
	
	local reactivelyPurgeSomeBinariesStream = Rx.ReplaySubject( 1 )
	
	leastRecentlyAccessedStream
	
		:tap( function( leastRecentlyAccessed, getNextBatch )
	
			maybePurgeLeastRecentlyAccessed( self, whatToPurge, willEventuallyPurgeProxies, spaceNeeded, 
					leastRecentlyAccessed, getNextBatch, purgeList, reactivelyPurgeSomeBinariesStream, maxSizePurgeable )
		
		end )
		
		:call( runUntilDone )
		
	return reactivelyPurgeSomeBinariesStream

end

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

local function getPurgeWindow( self, reason, whatToPurge, purgeWindowIndex )

	local dayIntervals = { 30, 21, 14, 7, 5, 3, 2, 1, 0 }

	if isProactivePurge( reason ) then
	
		-- During proactive purge we only purge Proxies that haven't been accessed
		-- in 60 days. Likewise, we only purge Originals that haven't been accessed
		-- within 15 days.
	
		if whatToPurge == 'proxy' then
			dayIntervals = { WFCatPhotoImportance._proxyLastAccessedThreshold }
		else
			dayIntervals = { WFCatPhotoImportance._originalLastAccessedThreshold }
		end
		
	end
	
	local beforeNDaysAgo = dayIntervals[ purgeWindowIndex ]
	
	local afterNDaysAgo
	
	if purgeWindowIndex > 1 then
		afterNDaysAgo = dayIntervals[ purgeWindowIndex - 1 ]
	end
	
	local isLastPurgeWindow = not dayIntervals[ purgeWindowIndex + 1 ]
	
	local purgeWindow = {
		beforeNDaysAgo = beforeNDaysAgo,
		afterNDaysAgo = afterNDaysAgo
	}
	
	return purgeWindow, isLastPurgeWindow
	
end

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

local function purgeToReclaimSpace( self, reason, purgeOriginals, purgeProxies, spaceNeeded, 
		purgeResultStream, purgeWindowIndex, purgeList )

	check( 'NmbVolumeAgent', 'number', 'boolean', 'boolean',  'number', '?table', '?number', '?table' )

	purgeOriginals = purgeOriginals or false
	purgeProxies = purgeProxies or false
	
	-- Maybe create a purge result stream which is used to signal to the caller that
	-- the purge is complete.
	
	purgeResultStream = purgeResultStream or Rx.ReplaySubject( 1 )
	
	-- Maybe create our purgeList, which we use to track assets that need to be purged.
	-- We first build up this list, and then purge the assets all at once.

	purgeList = purgeList or {
		proxies = {},
		originals = {}
	}
	
	-- We first purge assets that haven't been accessed within 30 days. If we need
	-- to reclaim more space, we shrink our purge window and try again.

	purgeWindowIndex = purgeWindowIndex or 1

	local whatToPurge = purgeOriginals and 'original' or purgeProxies and 'proxy'
		-- We always purge originals first, if both are requested.

	local maxSizePurgeable = whatToPurge == 'proxy' and self._maxSizeOfPurgeableProxies or nil

	if reason == reactivePurgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING then
		-- Purge beyond the idle threshold when purging to claim space for import/export/migration.
		maxSizePurgeable = nil
	end

	if self._debugDisableMaxSizePurgeableCheck then
		maxSizePurgeable = nil
	end
	
	local purgeWindow, isLastPurgeWindow = getPurgeWindow( self, reason, whatToPurge, purgeWindowIndex )
	
	logInfof( self, "Attempting to reclaim %.5f GB of disk space using %s that haven't"
		.. " been accessed between %d days and %d days.", 
		bytesToGb( spaceNeeded ), 
		purgeOriginals and 'originals' or 'proxies', 
		purgeWindow.beforeNDaysAgo,
		purgeWindow.afterNDaysAgo or math.huge )
		
	reactivelyPurgeSomeBinaries( self, whatToPurge, reason, purgeProxies, spaceNeeded, purgeWindow, purgeList, maxSizePurgeable )
		
		:flatMapLatest( function( purgeList, spaceNeeded )
		
			if spaceNeeded > 0 then

				logInfof( self, "Still need to reclaim %.5f GB of disk space.", bytesToGb( spaceNeeded ) )
				
				if not isLastPurgeWindow then
				
					-- We need to reclaim more space so shrink our purge window and try again.
					
					purgeWindowIndex = purgeWindowIndex + 1
					
					purgeToReclaimSpace( self, reason, purgeOriginals, purgeProxies, spaceNeeded, 
							purgeResultStream, purgeWindowIndex, purgeList )
							
					return Observable.empty()
					
				elseif purgeOriginals and purgeProxies then
				
					-- We still need to reclaim more space but we ran out of originals to purge.
					-- Reset the purge window, and start purging proxies.
					
					purgeToReclaimSpace( self, reason, false, purgeProxies, spaceNeeded, 
							purgeResultStream, nil, purgeList )
							
					return Observable.empty()
					
				end
				
			end

			local maybeDeleteAssets = Observable.just( true )			

			local allOrNothing = isAllOrNothing( reason )

			if not allOrNothing or 
				( allOrNothing and spaceNeeded <= 0 ) then

				maybeDeleteAssets = deleteOriginals( self, purgeList.originals )
				
					:merge( deleteProxies( self, purgeList.proxies ) )
					
					:takeLast( 1 )

			end
			
			local function completePurgeResultStream()
			
				purgeResultStream:onNext( spaceNeeded )
				purgeResultStream:onCompleted()
				
			end
			
			return maybeDeleteAssets:finally( completePurgeResultStream )
			
		end )
		
		:call( runUntilDone )

		return purgeResultStream

end

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

local function maybeSendDiskStats( self, freeSpace, totalSpace, sizeOfNimbusAssetsOnVolume )
	
	if not self._hasSentDiskStats and 
		freeSpace and 
		totalSpace and 
		sizeOfNimbusAssetsOnVolume then
	
		self._hasSentDiskStats = true

		local originalStorageLocationInfo = NmbFileManagementSettings.originalStorageLocationInfo()
		
		local broadcastData = {
			mainVolumeTargetDiskSpaceUsage = NmbFileManagementSettings.targetNimbusDiskSpacePercentage(),
			keepAllOriginals = NmbFileManagementSettings.keepAllOriginals(),
			keepAllProxies = NmbFileManagementSettings.keepAllProxies(),
			originalStorageLocation = originalStorageLocationInfo.readableIdentifier and 'custom' or 'default'
		}
		
		NmbTrackedDataBroadcaster.broadcastMessage( 'LocalStorageOptions', broadcastData )
		
		local diskStats = {
			sizeOfDrive = totalSpace,
			freeSpace = freeSpace,
			sizeOfNimbusAssetsOnVolume = sizeOfNimbusAssetsOnVolume,
		}
		
		NmbTrackedDataBroadcaster.broadcastMessage( 'NimbusDiskStats', diskStats )

	end

end

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

local function updateVolumeStatusStream( self )

	if self._volumeStatusStream then
	
		self._volumeStatusStream:onNext( {
			volumeGuid = self._volumeGuid,
			isOnline = self._isOnline, 
			isOverIdleSpaceThreshold = self._isOnline and self._isOutOfIdleSpace,
			isOutOfSpace = self._isOnline and self._isOutOfSpace,
			isOverTargetDiskSpaceUsage = self._isOnline and self._isOverTargetDiskSpaceUsage,
			isReadOnly = self._isReadOnly,
			remainingSpace = self._remainingSpace
		} )
		
	end
	
end

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

local function getFreeSpace( self, nimbusPath )

	-- Calculate how much free space is on the system.

	if self._debugMockDiskFreeSpace then
		return self._debugMockDiskFreeSpace
	end

	local success, fileSystemFreeSpace, totalSpace
	
	if nimbusPath then
		success, fileSystemFreeSpace, totalSpace = pcall( WFFileUtils.freeSpaceAtPath, nimbusPath )
	end
	
	fileSystemFreeSpace = success and fileSystemFreeSpace or nil
	
	if fileSystemFreeSpace then
		self._remainingSpace = fileSystemFreeSpace - self._minimumFreeSpace
		updateVolumeStatusStream( self )
	end
	
	return fileSystemFreeSpace, totalSpace

end

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

local function getSizeOfNegativesResultStream( self, nimbusPath )

	-- Returns an Observable that calculates the size of the local Negatives
	-- that sends one value and completes.

	local totalSpaceUsedByLocalNegatives = 0
	local totalSpaceUsedByLocalProxies = 0
	local totalSpaceUsedByLocalOriginals = 0

	if self._debugMockNimbusSize then

		totalSpaceUsedByLocalNegatives = self._debugMockNimbusSize
		totalSpaceUsedByLocalProxies = self._debugMockNimbusSize * 0.40
		totalSpaceUsedByLocalOriginals = self._debugMockNimbusSize * 0.60

		return Observable.just( totalSpaceUsedByLocalNegatives, totalSpaceUsedByLocalOriginals, totalSpaceUsedByLocalProxies )

	end

	return Observable.combineLatest( function( totalSpaceUsedByLocalProxies, totalSpaceUsedByLocalOriginals )

			totalSpaceUsedByLocalNegatives = totalSpaceUsedByLocalProxies + totalSpaceUsedByLocalOriginals

			logInfof( self, "Local proxies take up %.5f GB of space.", bytesToGb( totalSpaceUsedByLocalProxies ) )
			logInfof( self, "Local originals take up %.5f GB of space.", bytesToGb( totalSpaceUsedByLocalOriginals ) )
			logInfof( self, "In total, Local negatives take up %.5f GB of space.", bytesToGb( totalSpaceUsedByLocalNegatives ) )

			return totalSpaceUsedByLocalNegatives, totalSpaceUsedByLocalOriginals, totalSpaceUsedByLocalProxies

		end,
		WFAsyncFileUtils.calculatePathSize( child( nimbusPath, 'proxies' ) ),
		WFAsyncFileUtils.calculatePathSize( child( nimbusPath, 'originals' ) ) )

end

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

local function calculateMaxPurgeableSpace( binaryType, totalSpaceUsedByLocalBinaries, maxSpace )

	local maxPurgeableSpace = 0

	if binaryType == 'original' then
		
		local percentageToPurgeInThreshold = 1
		
		local spaceAboveToPurge = totalSpaceUsedByLocalBinaries - maxSpace
		local spaceToPurgeInThreshold = math.max( 0, maxSpace * percentageToPurgeInThreshold )
		
		maxPurgeableSpace = spaceAboveToPurge + spaceToPurgeInThreshold
		
	elseif binaryType == 'proxy' then

		local purgeAdjustment = 0.05
			-- Don't purge beyond idleSpaceThreshold or we may end up in a purge/idle download loop.

		local percentageToPurgeInThreshold = idleSpaceThreshold - purgeAdjustment

		local spaceAboveToPurge = totalSpaceUsedByLocalBinaries - maxSpace
		local spaceToPurgeInThreshold = math.max( 0, maxSpace * percentageToPurgeInThreshold )
		
		maxPurgeableSpace = spaceAboveToPurge + spaceToPurgeInThreshold

	end

	return maxPurgeableSpace

end

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

local function calculateTargetFreeSpaceResultStream( self, volumePath, fileSystemFreeSpace, targetDiskSpacePercentage )

	-- returns an Observable that sends one table value and then completes.

	return getSizeOfNegativesResultStream( self, volumePath )
		:map( function( totalSpaceUsedByLocalNegatives, totalSpaceUsedByLocalOriginals, totalSpaceUsedByLocalProxies )

			local targetFreeSpace = 0
			local percentOfTargetSpaceUsed = 0
			local percentOfTargetSpaceUsedByProxies = 0
			local percentOfTargetSpaceUsedByOriginals = 0

			local reservedSpace = self._minimumFreeSpace

			local maxSizeOfPurgeableProxies = 0
			local maxSizeOfPurgeableOriginals = 0

			if totalSpaceUsedByLocalNegatives then

				local totalFreeSpaceAndNimbusSpace = fileSystemFreeSpace + totalSpaceUsedByLocalNegatives - reservedSpace

				if self._keepAllOriginals and not self._keepAllProxies then

					local maxSpaceForProxies = math.max( 0, ( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalOriginals ) * targetDiskSpacePercentage )

					logInfof( self, "The max space local proxies should take up is %.5f GB", bytesToGb( maxSpaceForProxies ) )

					if maxSpaceForProxies == 0 then

						if totalSpaceUsedByLocalProxies > 0 then
							percentOfTargetSpaceUsedByProxies = math.huge
						else
							percentOfTargetSpaceUsedByProxies = 1
						end

					else
						percentOfTargetSpaceUsedByProxies = totalSpaceUsedByLocalProxies / maxSpaceForProxies
					end

					maxSizeOfPurgeableProxies = calculateMaxPurgeableSpace( 'proxy', totalSpaceUsedByLocalProxies, maxSpaceForProxies )
					maxSizeOfPurgeableOriginals = 0

					percentOfTargetSpaceUsed = percentOfTargetSpaceUsedByProxies

					targetFreeSpace = math.max( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalOriginals - maxSpaceForProxies, 0 )

				elseif self._keepAllProxies and not self._keepAllOriginals then

					local maxSpaceForOriginals = math.max( 0, ( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalProxies ) * targetDiskSpacePercentage )

					logInfof( self, "The max space local originals should take up is %.5f GB", bytesToGb( maxSpaceForOriginals ) )

					if maxSpaceForOriginals == 0 then

						if totalSpaceUsedByLocalOriginals > 0 then
							percentOfTargetSpaceUsedByOriginals = math.huge
						else
							percentOfTargetSpaceUsedByOriginals = 1
						end

					else
						percentOfTargetSpaceUsedByOriginals = totalSpaceUsedByLocalOriginals / maxSpaceForOriginals
					end

					maxSizeOfPurgeableProxies = 0
					maxSizeOfPurgeableOriginals = calculateMaxPurgeableSpace( 'original', totalSpaceUsedByLocalOriginals, maxSpaceForOriginals )

					percentOfTargetSpaceUsed = percentOfTargetSpaceUsedByOriginals

					targetFreeSpace = math.max( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalProxies - maxSpaceForOriginals, 0 )

				else

					local maxNimbusSpace = totalFreeSpaceAndNimbusSpace * targetDiskSpacePercentage

					logInfof( self, "The max space local negatives should take up is %.5f GB", bytesToGb( maxNimbusSpace ) )

					percentOfTargetSpaceUsed = totalSpaceUsedByLocalNegatives / maxNimbusSpace
					percentOfTargetSpaceUsedByOriginals = totalSpaceUsedByLocalOriginals / maxNimbusSpace
					percentOfTargetSpaceUsedByProxies = totalSpaceUsedByLocalProxies / maxNimbusSpace

					if self._keepAllProxies and self._keepAllOriginals then
						maxSizeOfPurgeableProxies = 0
						maxSizeOfPurgeableOriginals = 0
					else
						maxSizeOfPurgeableProxies = calculateMaxPurgeableSpace( 'proxy', totalSpaceUsedByLocalProxies, maxNimbusSpace )
						maxSizeOfPurgeableOriginals = calculateMaxPurgeableSpace( 'original', totalSpaceUsedByLocalOriginals, maxNimbusSpace )
					end

					targetFreeSpace = math.max( totalFreeSpaceAndNimbusSpace - maxNimbusSpace, 0 )

				end

			end

			targetFreeSpace = targetFreeSpace + reservedSpace

			return {
				targetFreeSpace = targetFreeSpace,
				totalSpaceUsedByLocalNegatives = totalSpaceUsedByLocalNegatives,
				percentOfTargetSpaceUsed = percentOfTargetSpaceUsed,
				percentOfTargetSpaceUsedByOriginals = percentOfTargetSpaceUsedByOriginals,
				percentOfTargetSpaceUsedByProxies = percentOfTargetSpaceUsedByProxies,
				maxSizeOfPurgeableOriginals = maxSizeOfPurgeableOriginals,
				maxSizeOfPurgeableProxies = maxSizeOfPurgeableProxies
			}

		end )

end

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

local function calculatePurgeBufferResultStream( self, reason )

	-- Returns an observable that sends the size of the purgeBuffer and completes.

	if reason == reactivePurgeReasons.PROACTIVE_ORIGINALS and self._keepAllOriginals then
		return Observable.just( 0 )
	elseif reason == reactivePurgeReasons.PROACTIVE_PROXIES and self._keepAllProxies then
		return Observable.just( 0 )
	elseif self._keepAllOriginals and self._keepAllProxies then
		return Observable.just( 0 )
	end
	
	local nimbusPath = buildAbsoluteNimbusPath( self )
	
	return getSizeOfNegativesResultStream( self, nimbusPath )
		:map( function( totalSpaceUsedByLocalNegatives, totalSpaceUsedByLocalOriginals, totalSpaceUsedByLocalProxies )

			local percentOfTargetSpaceUsed = 0

			local purgeBuffer = 0
			local maxSpace = 0

			if totalSpaceUsedByLocalNegatives then

				local fileSystemFreeSpace = getFreeSpace( self, nimbusPath )

				local totalFreeSpaceAndNimbusSpace = fileSystemFreeSpace - self._minimumFreeSpace + totalSpaceUsedByLocalNegatives

				if self._keepAllOriginals and not self._keepAllProxies then

					maxSpace = ( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalOriginals ) * self._targetDiskSpaceUsage
					percentOfTargetSpaceUsed = totalSpaceUsedByLocalProxies / maxSpace

				elseif self._keepAllProxies and not self._keepAllOriginals then

					maxSpace = ( totalFreeSpaceAndNimbusSpace - totalSpaceUsedByLocalProxies ) * self._targetDiskSpaceUsage
					percentOfTargetSpaceUsed = totalSpaceUsedByLocalOriginals / maxSpace

				else

					maxSpace = totalFreeSpaceAndNimbusSpace * self._targetDiskSpaceUsage
					percentOfTargetSpaceUsed = totalSpaceUsedByLocalNegatives / maxSpace

				end

			end

			if isProactivePurge( reason ) then

				if reason == reactivePurgeReasons.PROACTIVE_ORIGINALS then

					purgeBuffer = calculateMaxPurgeableSpace( 'original', totalSpaceUsedByLocalOriginals, maxSpace )

				elseif reason == reactivePurgeReasons.PROACTIVE_PROXIES then

					purgeBuffer = calculateMaxPurgeableSpace( 'proxy', totalSpaceUsedByLocalProxies, maxSpace )

				end

			elseif percentOfTargetSpaceUsed > 1 then

				purgeBuffer = maxSpace * reclaimTargetSpaceBuffer

			end

			if purgeBuffer > 0 then
				logInfof( self, "Using a purge buffer of %.5f GB.", bytesToGb( purgeBuffer ) )
			end

			return math.max( 0, purgeBuffer )

		end )

end

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

local function getDiskSpaceNeededResultStream( self, maybeExtraSpaceNeeded, hasClearedDeletionBacklog )

	-- Returns an Observable that sends one tuple containing the items below and completes.
	-- spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace, targetSpaceInfo, and fileSystemFreeSpace

	maybeExtraSpaceNeeded = maybeExtraSpaceNeeded or 0

	-- Calculate how much free space is on the system.

	local nimbusPath = buildAbsoluteNimbusPath( self )

	local fileSystemFreeSpace, totalSpace = getFreeSpace( self, nimbusPath )

	if fileSystemFreeSpace == nil then

		self._maxSizeOfPurgeableProxies = nil

		logWarnf( self, "Unable to get free space on path %s", nimbusPath )

		return Observable.just( 0, 0, {}, 0 )

	end

	-- If Nimbus breaches the noDiskSpaceThreshold value, then Nimbus is in a no-disk-space state.
	-- In this scenario, we try to purge originals and smart previews.

	local noDiskSpaceThreshold = self._minimumFreeSpace

	-- Calculate how much free space we should try to keep on the system to reach the target disk space usage.

	return calculateTargetFreeSpaceResultStream( self, nimbusPath, fileSystemFreeSpace, self._targetDiskSpaceUsage )
		:flatMap( function( targetSpaceInfo )

			local spaceNeededToPreventNoDiskSpace = math.max( 0,
				noDiskSpaceThreshold - fileSystemFreeSpace + maybeExtraSpaceNeeded )

			local spaceNeededToReachTargetDiskSpace = math.max( 0,
				targetSpaceInfo.targetFreeSpace - fileSystemFreeSpace + maybeExtraSpaceNeeded - spaceNeededToPreventNoDiskSpace )

			-- Debug overrides used for only unit tests
			if self._debugSpaceNeededToReachTargetDiskSpace then
				spaceNeededToReachTargetDiskSpace = self._debugSpaceNeededToReachTargetDiskSpace
			end
			if self._debugSpaceNeededToPreventNoDiskSpace then
				spaceNeededToPreventNoDiskSpace = self._debugSpaceNeededToPreventNoDiskSpace
			end

			maybeSendDiskStats( self, fileSystemFreeSpace, totalSpace, targetSpaceInfo.totalSpaceUsedByLocalNegatives )

			if not hasClearedDeletionBacklog and
				( spaceNeededToPreventNoDiskSpace > 0 or spaceNeededToReachTargetDiskSpace > 0 ) then
		
				-- If we detect we need to reclaim space ( reactive purge ), first process all the deletions
				-- in the backlog and then check again.
		
				return self._deletionBacklog:processDeletionBacklog()
					:flatMapLatest( function()
						return getDiskSpaceNeededResultStream( self, maybeExtraSpaceNeeded, true )
					end )

			end

			self._maxSizeOfPurgeableProxies = targetSpaceInfo.maxSizeOfPurgeableProxies

			return Observable.just(
					spaceNeededToPreventNoDiskSpace,
					spaceNeededToReachTargetDiskSpace,
					targetSpaceInfo,
					fileSystemFreeSpace or 0 )

		end )

end

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

local function isVolumeOnlineStream( self )

	return Observable.timer( 0, volumeStatusCheckPeriod, WFRxScheduler.sharedScheduler() )
	
		:map( function()
		
			local isOnline = WFFileUtils.isVolumeGuidOnline( self._volumeGuid )

			local isReadOnly = isOnline and WFFileUtils.isVolumeReadOnly( self._volumeGuid )

			return isOnline, isReadOnly
			
		end )
	
		:distinctUntilChanged()
	
		:map( function( isOnline, isReadOnly )
		
			if isOnline then
				logInfof( self, "Volume %s is now online!", self._volumeGuid )
			else
				logInfof( self, "Volume %s is now offline!", self._volumeGuid )
			end
			
			if isReadOnly then
				logInfof( self, "Volume %s is read only!", self._volumeGuid )
			else
				logInfof( self, "Volume %s is not read only!", self._volumeGuid )
			end
			
			if isReadOnly then
				isOnline = false
			end
			
			return isOnline, isReadOnly
		
		end )
	
end

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

-- spaceNeededStream fires periodically every time the system detects that it needs
-- to reclaim space.

local function spaceNeededStream( self )

	local reactivePurgeDelayOnStartup = self._debugReactivePurge and 0 or 10
		-- Delay reactive purge for 10 seconds on startup
		
	return Observable.timer( reactivePurgeDelayOnStartup, WFRxScheduler.sharedScheduler() )
	
		:flatMapLatest( function()
		
			-- Trigger purge based on the doc store change counter.
			return self._docStore:changeCounter()
				
		end )
	
		:merge( Observable.timer( diskSpaceCheckPeriod, diskSpaceCheckPeriod, WFRxScheduler.sharedScheduler() ) )
			-- Periodically do a disk space check, regardless of the doc store change counter.
			
		:calm( self._maxDiskSpacePollingInterval, WFRxScheduler.sharedScheduler() )
			-- Limit how often we do a disk space check. This is to prevent us from checking
			-- every time the change counter fires.
			
		:flatMapLatest( function()
		
			logInfof( self, "Checking if we need to reclaim disk space.." )
			return getDiskSpaceNeededResultStream( self )

		end )
		
		:filter( function( 
					spaceNeededToPreventNoDiskSpace, 
					spaceNeededToReachTargetDiskSpace,
					targetSpaceInfo,
					currentFreeSpace )

			-- spaceNeededToPreventNoDiskSpace is the amount of space needed
			-- to prevent a no disk space scenario.
			
			-- spaceNeededToReachTargetDiskSpace is the amount of space needed
			-- to reach the target disk space usage, minus spaceNeededToPreventNoDiskSpace.

			local shouldContinue = false
			
			-- Check if we are purging because we ran out of disk space.
			
			if spaceNeededToPreventNoDiskSpace > 0 then
				shouldContinue = true
			else
			
				logInfof( self, "We're NOT out of space. This is good!" )
				self._isOutOfSpace = false
			
			end
			
			-- Check if we are purging because we want to reach our target disk space.
			
			if spaceNeededToReachTargetDiskSpace > 0 then 
				shouldContinue = true
			elseif not shouldContinue then
			
				logInfof( self, "We're within our target disk space usage. This is good!" )
				self._isOverTargetDiskSpaceUsage = false
				
			end
		
			-- Check if we're out of of "idle" space. AKA space reserved for Smart Preview
			-- downloads by the idle downloader.
			
			local percentOfTargetSpaceUsedByProxies = targetSpaceInfo.percentOfTargetSpaceUsedByProxies or 0
			
			if percentOfTargetSpaceUsedByProxies < idleSpaceThreshold then 
				logInfof( self, "We're NOT out of idle space." )
				self._isOutOfIdleSpace = false
			else
				logInfof( self, "We're out of idle space." )
				self._isOutOfIdleSpace = true
			end
		
			updateVolumeStatusStream( self )
			
			return shouldContinue
		
		end )
		
		:calm( reactivePurgeCalmInterval, WFRxScheduler.sharedScheduler() )
			-- Further calm how often reactive purge can run.
		
end

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

local function reclaimSpace( self, purgeReason, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )

	self._reactivePurgeInProgress = true
	self._reactivePurgeInProgressStream:onNext( self._reactivePurgeInProgress )

	if spaceNeededToPreventNoDiskSpace > 0 then
		logInfof( self, "Need to reclaim %.5f GB of space to prevent a NO disk space warning.", 
				bytesToGb( spaceNeededToPreventNoDiskSpace ) )
	end
	
	if spaceNeededToReachTargetDiskSpace > 0 then
	
		if isProactivePurge( purgeReason ) then
			logInfof( self, "Going to try to reclaim %.5f GB of space via proactive purge!", 
					bytesToGb( spaceNeededToReachTargetDiskSpace ) )
		else
			logInfof( self, "Need to reclaim %.5f GB of space to reach our target disk space usage.", 
					bytesToGb( spaceNeededToReachTargetDiskSpace ) )
		end
		
	end
	
	self._isOriginalMarkedForPurgeByLocalId = {}
	self._isProxyMarkedForPurgeByLocalId = {}
		-- Clear the caches of previously purged originals and proxies.
	
	-- 1.) Purge assets to prevent a no-disk-space scenario. This might be a no-op.
	
	purgeReason = purgeReason or reactivePurgeReasons.NO_DISK_SPACE
	
	local purgeOriginals = not self._keepAllOriginals
	local purgeProxies = not self._keepAllProxies
	
	if purgeReason == reactivePurgeReasons.PROACTIVE_ORIGINALS then
		purgeProxies = false
	elseif purgeReason == reactivePurgeReasons.PROACTIVE_PROXIES then
		purgeOriginals = false
	end
	
	local purgeResultStream
	
	if not purgeOriginals and not purgeProxies then
	
		purgeResultStream = Observable.just( spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
		
	else
	
		local amountOfSpaceToPurge = spaceNeededToPreventNoDiskSpace
		
		local maybePurgeToReclaimSpace = Observable.just( amountOfSpaceToPurge )
		
		if amountOfSpaceToPurge > 0 then
		
			logInfof( self, "Attempting to reclaim %.5f GB of space to prevent a NO DISK SPACE scenario.", 
					bytesToGb( amountOfSpaceToPurge ) )
					
			maybePurgeToReclaimSpace = purgeToReclaimSpace( self, purgeReason, purgeOriginals, purgeProxies, amountOfSpaceToPurge )
					
		end

		purgeResultStream = maybePurgeToReclaimSpace

			:flatMapLatest( function( spaceNeededAfterPurgeToPreventNoDiskSpace )

				if purgeReason ~= reactivePurgeReasons.PROACTIVE_PROXIES
					and purgeReason ~= reactivePurgeReasons.PROACTIVE_ORIGINALS
					and purgeReason ~= reactivePurgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING then

					return Observable.combineLatest(
						Observable.just( spaceNeededAfterPurgeToPreventNoDiskSpace ),
						calculatePurgeBufferResultStream( self ) )

				end

				return Observable.just( spaceNeededAfterPurgeToPreventNoDiskSpace, 0 )

			end )

			:flatMapLatest( function( spaceNeededAfterPurgeToPreventNoDiskSpace, newPurgeBuffer )
			
				if spaceNeededAfterPurgeToPreventNoDiskSpace <= 0 then

					spaceNeededToReachTargetDiskSpace = spaceNeededToReachTargetDiskSpace + spaceNeededAfterPurgeToPreventNoDiskSpace

					-- 2.) Purge assets to reach the target disk space threshold. This might be a no-op.

					amountOfSpaceToPurge = spaceNeededToReachTargetDiskSpace + newPurgeBuffer
					
					if isProactivePurge( purgeReason ) then
						logInfof( self, "Now trying to reclaim %.5f GB of space via proactive purge!", 
								bytesToGb( amountOfSpaceToPurge ) )
					else
						logInfof( self, "Attempting to reclaim %.5f GB of space to reach our target disk space usage.", 
								bytesToGb( amountOfSpaceToPurge ) )
								
					end
					
					purgeReason = purgeReason or reactivePurgeReasons.TARGET_RANGE
					
					return purgeToReclaimSpace( self, purgeReason, purgeOriginals, purgeProxies, amountOfSpaceToPurge )
					
						:map( function( value )
							return value - newPurgeBuffer
						end )

						:map( function( spaceNeededAfterPurgeToReachTargetDiskSpace )
						
							return spaceNeededAfterPurgeToPreventNoDiskSpace, 
									spaceNeededAfterPurgeToReachTargetDiskSpace
						
						end )
				
				end 
			
				-- No-disk-space purge failed.
				
				return Observable.just( 
						spaceNeededAfterPurgeToPreventNoDiskSpace,
						spaceNeededToReachTargetDiskSpace )
			
			end )
		
	end
		
	return purgeResultStream
	
		:finally( function()
		
			self._reactivePurgeInProgress = false
			self._reactivePurgeInProgressStream:onNext( self._reactivePurgeInProgress )
			
		end )
		
end

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

local function reclaimSpaceAndReport( self, purgeReason, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace, operationStartTime )
	
	local startTime = KSDate.timeNowForAnimation()
	
	return reclaimSpace( self, purgeReason, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
		
		:tap( function( spaceNeededAfterPurgeToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
			
			local endTime = KSDate.timeNowForAnimation()
			
			local reason
			if purgeReason == reactivePurgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING then
				reason = "reactive-on-demand"
			elseif purgeReason == reactivePurgeReasons.PROACTIVE_ORIGINALS then
				reason = "proactive-originals"
			elseif purgeReason == reactivePurgeReasons.PROACTIVE_PROXIES then
				reason = "proactive-proxies"
			else
				reason = "reactive"
			end

			local broadcastData = {
				reason = reason,
				spaceNeededToPreventNoDiskSpace = spaceNeededToPreventNoDiskSpace,
				spaceNeededToReachTargetDiskSpace = spaceNeededToReachTargetDiskSpace,
				spaceNeededAfterPurgeToPreventNoDiskSpace = spaceNeededAfterPurgeToPreventNoDiskSpace,
				spaceNeededAfterPurgeToReachTargetDiskSpace = spaceNeededToReachTargetDiskSpace,
				timeTakenForPurge = endTime - startTime,
				timeTakenIncludingDiskCheck = operationStartTime and ( endTime - operationStartTime ) or nil,
			}
			
			NmbTrackedDataBroadcaster.broadcastMessage( 'PurgeStats', broadcastData )
			
			logInfof( self, "Reclaim space for %s took %s", reason, tostring( endTime - startTime ) )
			
			if operationStartTime then
				logInfof( self, "Reclaim space for %s took %s including disk check", reason, tostring( endTime - operationStartTime ) )
			end
			
		end )
end

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

local function ongoingReactivePurge( self )

	return isVolumeOnlineStream( self )

		:flatMapLatest( spaceNeededStream( self ) )
	
		:flatMap( function( 
					spaceNeededToPreventNoDiskSpace, 
					spaceNeededToReachTargetDiskSpace )
					
			return self._reactivePurgeTask:throwIfNotRunning( spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
				-- Need to throw as this function doesn't ever complete.
				
		end )
	
		:flatMap( function( 
					spaceNeededToPreventNoDiskSpace, 
					spaceNeededToReachTargetDiskSpace )
					
			if self._reactivePurgeInProgress then
			
				logInfof( self, "Reactive purge already in progress." )	
				return Observable.empty()
				
			else
			
				return reclaimSpaceAndReport( self, false, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
			
			end
			
		end )

		:tap( function( 
					spaceNeededAfterPurgeToPreventNoDiskSpace, 
					spaceNeededAfterPurgeToReachTargetDiskSpace )
		
			if spaceNeededAfterPurgeToPreventNoDiskSpace > 0 then
			
				if not self._isOutOfSpace then
				
					local broadcastData = {					
						description = "Reactive purge failed. Now out of disk space.",
						spaceNeededAfterPurge = spaceNeededAfterPurgeToPreventNoDiskSpace,
					}
				
					NmbTrackedDataBroadcaster.broadcastMessage( 'ReactivePurgeFailed', broadcastData )
					
				end
				
				self._isOutOfSpace = true
				self._isOverTargetDiskSpaceUsage = true
				
				logWarnf( self, "Unable to reclaim needed disk space. Still need %.5f GB.", bytesToGb( spaceNeededAfterPurgeToPreventNoDiskSpace ) )
				logWarnf( self, "We are now in a NO DISK SPACE state." )
				
				
			
			elseif spaceNeededAfterPurgeToReachTargetDiskSpace > 0 then
			
				self._isOutOfSpace = false
				self._isOverTargetDiskSpaceUsage = true

				logInfof( self, "Unable to achieve target disk space. Still need %.5f GB.", bytesToGb( spaceNeededAfterPurgeToReachTargetDiskSpace ) )
			
			else
			
				self._isOutOfSpace = false
				self._isOverTargetDiskSpaceUsage = false

				logInfof( self, "Needed space has been successfully reclaimed." )
				
			end
		
			updateVolumeStatusStream( self )
		
			self._reactivePurgeRunComplete:onNext( true )
		
		end,
		function() 
			updateVolumeStatusStream( self )
		end,
		function() 
			updateVolumeStatusStream( self )
		end )

		:catch( function( errorMessage )
		
			self._reactivePurgeInProgress = false
			self._reactivePurgeInProgressStream:onNext( self._reactivePurgeInProgress )
			
			return self._reactivePurgeTask:throwIfNotStopped( errorMessage )
			
		end )

end

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

local daysToSeconds = function( days )
	return days * 24 * 60 * 60
end

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

local function ongoingProactivePurge( self )

	local proactivePurgeDelayOnStartup = self._debugProactivePurge and 0.01 or 30

	return Observable.timer( proactivePurgeDelayOnStartup, WFRxScheduler.sharedScheduler() )
		 -- Delay on startup
			
		:take( 1 )
		
		:flatMap( Observable.timer( 0, self._debugProactivePurge and 10 or daysToSeconds( 1 ), WFRxScheduler.sharedScheduler() ) )
			-- Every 24 hours, run proactive purge
			
		:flatMapLatest( function()
			return self._proactivePurgeTask:throwIfNotRunning()
				-- Need to throw as this function doesn't ever complete.
		end )
		
		:flatMapLatest( function()
		
			-- Wait until the volume is online, and no reactive purge is running.
		
			return Observable.combineLatest( self._reactivePurgeInProgressStream, isVolumeOnlineStream( self ) )
	
				:filter( function( reactivePurgeInProgress, isVolumeOnline )
				
					if isVolumeOnline and not self._reactivePurgeInProgress then
					
						return true
					
					end
					
					logInfof( self, "Can't yet proactively purge originals. Waiting..." )	
					
					return false
					
				end )
				
				:take( 1 )
	
		end )
		
		:flatMapLatest( function()

			if not self._keepAllOriginals then
			
				logInfof( self, "Starting proactive purge for Originals!" )	
			
				local reason = reactivePurgeReasons.PROACTIVE_ORIGINALS
			
				return calculatePurgeBufferResultStream( self, reason )
					:flatMapLatest( function( spaceToAttemptToReclaimProactively )

						if spaceToAttemptToReclaimProactively > 0 then
							return reclaimSpaceAndReport( self, reason, 0, spaceToAttemptToReclaimProactively )
						end

						return Observable.just( true )

					end )

			end
			
			return Observable.just( true )
			
		end )

		:flatMapLatest( function()
		
			-- Wait until the volume is online, and no reactive purge is running.
		
			return Observable.combineLatest( self._reactivePurgeInProgressStream, isVolumeOnlineStream( self ) )
	
				:filter( function( reactivePurgeInProgress, isVolumeOnline )
				
					if isVolumeOnline and not self._reactivePurgeInProgress then
					
						return true
					
					end
					
					logInfof( self, "Can't yet proactively purge proxies. Waiting..." )	
					
					return false
					
				end )
				
				:take( 1 )
	
		end )

		:flatMapLatest( function()
		
			if not self._keepAllProxies then
			
				logInfof( self, "Starting proactive purge Proxies!" )	
			
				local reason = reactivePurgeReasons.PROACTIVE_PROXIES
			
				return calculatePurgeBufferResultStream( self, reason )
					:flatMapLatest( function( spaceToAttemptToReclaimProactively )

						if spaceToAttemptToReclaimProactively > 0 then
							return reclaimSpaceAndReport( self, reason, 0, spaceToAttemptToReclaimProactively )
						end

						return Observable.just( true )

					end )

			end
			
			return Observable.just( true )
			
		end )

		:tap( function()
		
			self._proactivePurgeRunComplete:onNext( true )
		
		end )

		:catch( function( errorMessage )
			self._proactivePurgeRunComplete:onNext( true )
			return self._proactivePurgeTask:throwIfNotStopped( errorMessage )
		end )

end

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

local function setupPurgeOnDelete( self )

	-- When as asset is deleted from the doc store, we immediately purge all
	-- files associated with that asset.

	local purgeAgentChangeCounterLocalDocId = 'NmbVolumeAgent-changeCounter'
	
	self._purgeOnDelete = self._docStore:deletedDocsSinceNamedChangeCounter( 
			purgeAgentChangeCounterLocalDocId, 
			{
				_localOnly = {
					files = true,
					deletedLocally = true
				},
				_localId = true,
				_serverProvidedMetadata = {
					new_id = true,
				},
				type = true
			} )

		:flatMap( function( docValues, continueFrom )
			
			for _, docValue in ipairs( docValues ) do
			
				-- deleted assets from Oz have no type information by this point, so
				-- pass those through for deletion just in case they were assets, but
				-- we might as well not delete things that we _know_ were not assets
			
				-- If we sync down a deleted version document, the associated binaries
				-- are cleaned up automatically by PixelProviderFileCachePurgeWorker.
			
				if docValue and ( docValue.type == "asset" or docValue.type == nil ) then
				
					self._fileAccessDb:delete( docValue._localId )
						-- Remove entry from NmbFileAccessDatabase
						
					local newId = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'new_id' )
					local localOnly = KSTableUtils_derefOpt( docValue, "_localOnly" )
						
					-- local bulk asset deletion takes care of file deletion (because it
					-- is faster to do it in batches), but if we sync asset deletion from Oz,
					-- we need to take care of the items as they arrive.

					-- If the asset has a newId then oz is has moved it to trash or
					-- restored it, don't delete the binaries as the asset will still
					-- refer to it.

					if newId == nil and type( localOnly ) == 'table' and not localOnly.deletedLocally then

						local fileArray = KSTableUtils_derefOpt( localOnly, "files" )
						
						if type( fileArray ) == 'table' then
						
							WFCatDelete.deleteDocStoreFileArray( fileArray, self._deletionBacklog )
						
						end
						
					end
				
				end
			
			end
			
			return self._docStore:updateDeletedDocsSinceNamedChangeCounter(
					purgeAgentChangeCounterLocalDocId, continueFrom )
			
		end )
		
		:subscribe()

end 

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

--[[!

	\brief Calculates how much available space is on the volume.
	\param volume (string) The path of the volume.
	\return (number) amount of free space in bytes.

--]]

local function checkAvailableSpaceOnVolume( self, volume )

	check( 'table', 'string' )

	local success, freeSpaceOnVolume = pcall( WFFileUtils.freeSpaceAtPath, volume )

	if not success then
		return 0
	end
			
	return math.max( 0, freeSpaceOnVolume - NmbFileManagementSettings.minimumFreeSpace() )

end

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

--[[

	NmbVolumeAgent; A Guide.

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

	For each separate volume, we create an instance of NmbVolumeAgent. NmbVolumeAgent
	is used to monitor the volume disk space, connection status, and to purge
	assets to reclaim disk space.

	There is a distinction between purging originals/proxies and purging 
	renditions/previews. NmbVolumeAgent is responsible for the former and takes
	into account a number of variables when considering what binaries are purgeable.
	Purging of renditions and previews is handled by the pixel provider.
	
	It should be noted that NmbVolumeAgent also handles the purging of binaries
	from disk when an asset is deleted from Nimbus. This process is unrelated
	to the ongoing purge task. See _purgeOnDelete. 

	Purge operates differently depending on the type of volume, the
	target disk space threshold, and what's being stored on the volume.
		
	----------------------------------------------------------------------------
		
	Primary Volume
		
		N% Target Disk Space Usage ( Automatic ):
		
		Nimbus manages disk space on the primary volume by purging originals and
		proxies from disk when we detect they are no longer required by the user. 
		
		Keep All Originals:
	
		With this option enabled, Nimbus will never remove originals from disk unless they
		are explicitly deleted by the user.
		
		Keep All Proxies:
		
		With this option enabled, Nimbus will never remove proxies from disk unless they
		are explicitly deleted by the user.
		
	Secondary Volume
		
		If a secondary volume is being used to store negatives, we only start purging negatives
		proactively.
		
		Keep All Originals:
	
		With this option enabled and a secondary volume selected, Nimbus will never remove 
		originals from disk unless theyare explicitly deleted by the user.
		
		Keep All Proxies:
		
		With this option enabled and a secondary volume selected, Nimbus will never remove 
		proxies from disk unless theyare explicitly deleted by the user. 

	----------------------------------------------------------------------------
	
	We use a concept of importance when deciding when to proactively purge an asset.
	
	A photo is considered 'important' or 'active' if the following conditions are met.
		
		- Photo is flagged as a "Pick".
		- Photo is rated at or above n stars. n defaults to 4.
		- Photo has been modified ( metadata or develop settings ) within n days.
			n defaults to 15.
		- Photo was imported within n days. n defaults to 15.
		- Photo is marked for Offline Use
		
	We proactively purge some of the photos that are not 'active'.
		
	Further requirements for Original purging:
		- Oz has a copy of the Original.
		- Photo was accessed within 15 days.
		
	Further requirements for Proxy purging:
		- Oz has a copy of the Original or Proxy.
		- Photo was accessed within 60 days.
		
	----------------------------------------------------------------------------

	When to Purge and What to Purge:

	Regardless of the mode you're in, the volume agent will remove binaries both proactively
	and reactively.
	
	Proactive Purge - Remove some 'inactive' photos once a day. This happens shortly after the
	app starts up, and then repeats every 24 hours.
	
	Reactive Purge - Remove binaries from the volume when Nimbus detects that the volume
	is over the target disk space usage.
	
	We also take recent access time into consideration when purging. We purge the least
	recently accessed photos first. A photo is considered `accessed` if develop settings
	are retrieved for the photo, which indicates the user viewed the image in loupe or
	develop.
	
	For more info, see: https://wiki.corp.adobe.com/display/lightroom/LrCC+Local+Storage+Rules+for+1.1
		
]]--

local NmbVolumeAgent = {

	_logger = false,
		-- (WFLogger) A logger instance for the volume agent. There's one logger per
		-- volume, instead of having a global logger for the file.

	_docStore = false,
		-- (WFRxDocStore)

	_indexStore = false,
		-- (WFRxIndexStore)

	_idleDownloader = false,
		-- (WFCatIdleDownloader)

	_fileAccessDb = false,
		-- (NmbFileAccessDatabase)

	_deletionBacklog = false,
		-- (NmbMRUDatabase)

	_nimbusPath = false,
		-- (string) The partial path to where Nimbus assets are stored on the volume.

	_volumeGuid = false,
		-- (string) The unique guid of the volume.

	_keepAllOriginals = false,
		-- (boolean) If true, the user is choosing to keep all originals local.
		--	This means that we won't ever purge an original.

	_keepAllProxies = false,
		-- (boolean) If true, the user is choosing to keep all proxies local.
		-- This means that we won't ever purge a proxy.

	_targetDiskSpaceUsage = false,
		-- (number) The target percent of disk space that Nimbus tries to use.

	_maxSizeOfPurgeableProxies = false,
		-- (number) The total size of purgeable proxies ( proxies consuming non-idle space ).

	_reactivePurgeTask = false,
		-- (WFRxContinuousTask) A continuous task to handle reactive purging.
	
	_proactivePurgeTask = false,
		-- (WFRxContinuousTask) A continuous task to handle proactive purging.

	_reactivePurgeInProgress = false,
		-- (boolean) If true, there is a reactive purge in progress.

	_reactivePurgeInProgressStream = false,
		-- (Rx.BehaviorSubject) Fires when there's a change to _reactivePurgeInProgress.

	_purgeOnDelete = false,
		-- (Rx Subscription) A subscription to an Rx stream that purges
		-- files from disk when the associated asset is deleted from the
		-- doc store.

	_maxDiskSpacePollingInterval = false,
		-- (number) recalculate disk space needs, at most, every N seconds.
		
	_minimumFreeSpace = false,
		-- (number) The very minimum free space, in bytes.
		
	_volumeStatusStream = false,
		-- (Rx.ReplaySubject) A stream that fires with info about the status of the volume.
		
	_isSyncPausedStream = false,
		-- (Rx.BehaviorSubject) A stream that fires when global sync is paused or resumed.
		
	_hasInternetConnectionStream = false, 
		-- (Rx.BehaviorSubject) A stream that fires when the internet goes offline or online.
		
	_isOverTargetDiskSpaceUsage = false,
		-- (boolean) Track whether or not we are over the target disk space usage.
	
	_isOutOfIdleSpace = false,
		-- (boolean) Track whether or not we have room on this volume for idle downloads.
	
	_isOutOfSpace = false,
		-- (boolean) Track whether or not we have no disk space.
		
	_isOnline = false,
		-- (boolean) Track whether or not the disk is online.
		
	_isReadOnly = false,
		-- (boolean) Track whether or not the volume is read-only.
		
	_skipPurgeOfNMostRecentAssets = false,
		-- (number) The number of most recently accessed assets to prevent from being purged.
		
	_protectedFromPurge = false,
		-- (table) A list of assets to protect from being purged.
		-- key: asset id
		-- value: ref count 
		
	_volumeConnectionSubscription = false,
		-- (Rx Subscription) A subscription to an obesrvable that periodically checks the volumes connection status.
		
	_reactivePurgeRunComplete = false,
		-- (Rx Subject) Fires each time a reactive purge run completes.
		
	_proactivePurgeRunComplete = false,
		-- (Rx Subject) Fires each time a proactive purge run completes.
		
	_hasSentDiskStats = false,
		-- (bolean)
		
	_remainingSpace = false,
		-- (number) The amount of remaining free space on the volume, not including the reserved space.
		
	_isOriginalMarkedForPurgeByLocalId = false,
	
	_isProxyMarkedForPurgeByLocalId = false,
		
	_debugSpaceNeededToReachTargetDiskSpace = false,
		-- (number)
		
	_debugSpaceNeededToPreventNoDiskSpace = false,
		-- (number)

	_debugProactivePurge = false,
		-- (boolean) If true, the volume agent proactive purge will trigger right away. Used in testing.

	_debugReactivePurge = false,
		-- (boolean) If true, the volume agent reactive purge will start without delay. Used in testing.

	_debugMockDiskFreeSpace = false,
		-- (number) If set, mocks the total free space.
		
	_debugMockNimbusSize = false,
		-- (number) If set, mocks the total size of Nimbus.
		
	_debugDisableMaxSizePurgeableCheck = false,
		-- (boolean) If set, purge ignores the max purgeable size check.
		
	_debugLogRingBuffer = false,
		-- (WFRingBuffer) A ring buffer of logging output that is included in the diagnostic log.
		
	WFCloseSupport.addCloseSupportMethods,

	private = {
		
		-- The following are methods are internal, but are exposed for testing.
		
		_getDiskSpaceNeededResultStream = getDiskSpaceNeededResultStream,
		_calculateTargetFreeSpaceResultStream = calculateTargetFreeSpaceResultStream,
		_calculatePurgeBufferResultStream = calculatePurgeBufferResultStream,
		_purgeToReclaimSpace = purgeToReclaimSpace,
		_reactivelyPurgeSomeBinaries = reactivelyPurgeSomeBinaries,
		_spaceNeededStream = spaceNeededStream,
		
	},
	
	class = {
		checkAvailableSpaceOnVolume = checkAvailableSpaceOnVolume
	}
	
}

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

function NmbVolumeAgent:__init( catalog, idleDownloader, volumeGuid, nimbusPath,
								targetDiskSpaceUsage, keptLocalInfo, optOverrides )

	check( 'NmbVolumeAgent', 'WFCatalog', '?WFCatIdleDownloader', '+string', 'string', 'number', 'table', '?table' )

	self._docStore = catalog:docStore()

	self._indexStore = catalog:indexStore()

	self._fileAccessDb = catalog:fileAccessDb()
	
	self._idleDownloader = idleDownloader
	
	self._deletionBacklog = catalog:getDeletionBacklog()

	self._volumeGuid = volumeGuid

	self._nimbusPath = nimbusPath
	
	self._logger = WFLogger( 'NmbVolumeAgent:' .. self._volumeGuid )
	
	self._debugLogRingBuffer = WFRingBuffer.make( maxDebugLogSize )
	
	self._targetDiskSpaceUsage = math.max( targetDiskSpaceUsage, 0.0001 )
		-- If the target space is 0, use a very, very small cache.
		
	self._keepAllOriginals = keptLocalInfo.keepAllOriginals

	self._keepAllProxies = keptLocalInfo.keepAllProxies

	self._volumeStatusStream = Rx.ReplaySubject( 1 )
	
	self._isSyncPausedStream = Rx.BehaviorSubject()
	
	self._hasInternetConnectionStream = Rx.BehaviorSubject()
	
	self._reactivePurgeRunComplete = Rx.Subject()
	
	self._proactivePurgeRunComplete = Rx.Subject()
	
	self._protectedFromPurge = {}
	
	self._isOriginalMarkedForPurgeByLocalId = {}
	
	self._isProxyMarkedForPurgeByLocalId = {}
	
	self._reactivePurgeInProgressStream = Rx.BehaviorSubject( false )	

	self._remainingSpace = 0

	optOverrides = optOverrides or {}
	
	-- Purge settings
	
	self._maxDiskSpacePollingInterval = optOverrides.diskSpacePollingInterval or NmbFileManagementSettings.maxPollingInterval()
	self._minimumFreeSpace = optOverrides.minimumFreeSpace or NmbFileManagementSettings.minimumFreeSpace()
	self._skipPurgeOfNMostRecentAssets = optOverrides.skipPurgeOfNMostRecentAssets or WFConfiguration.Nimbus.skipPurgeOfNMostRecentAssets or 5
	
	-- Purge debug overrides
	
	self._debugSpaceNeededToReachTargetDiskSpace = optOverrides.debugSpaceNeededToReachTargetDiskSpace or false
	self._debugSpaceNeededToPreventNoDiskSpace = optOverrides.debugSpaceNeededToPreventNoDiskSpace or false
	self._debugProactivePurge = optOverrides.debugProactivePurge or false
	self._debugReactivePurge = optOverrides.debugReactivePurge or false
	self._debugMockDiskFreeSpace = optOverrides.debugMockDiskFreeSpace or false
	self._debugMockNimbusSize = optOverrides.debugMockNimbusSize or false
	self._debugDisableMaxSizePurgeableCheck = optOverrides.debugDisableMaxSizePurgeableCheck or false
	
	-- Delete the temp folder. Each per-volume temp folder is used as a staging ground for
	-- PP downloads. Any asset that exists within it on startup is safe to delete.

	-- TODO [jandrew 2017-05-05]: Instead of deleting the temp folder on startup, we
	-- may want to eventually delete the contents of the temp folder based on when
	-- they were created. We could delete assets from the temp folder after they've
	-- been around for more than a few days. Coupling this with logic in the PP to
	-- check for an asset within the temp folder before downloading, could prevent
	-- us from redownloading assets that were already downloaded but then abandoned.
	
	local tempFolder = catalog:fileSpec():tempFolderPathForVolume( self._volumeGuid )
	
	if KSPathUtils.getStandardFilePath( "temp" ) ~= tempFolder then
	
		local success = pcall( WFFileUtils.delete, tempFolder )
		
		if not success then
		
			logInfof( self, "Unable to delete temp folder %s on volume %s", tempFolder, self._volumeGuid )
		
		end
	
	end

	-- Monitor Volume Connection
	
	self._volumeConnectionSubscription = isVolumeOnlineStream( self )
		:subscribe( function( isOnline, isReadOnly )
			self._isOnline = isOnline
			self._isReadOnly = isReadOnly
			updateVolumeStatusStream( self )
		end )
	
	-- Purge on Delete
	
	setupPurgeOnDelete( self )

	-- Reactive Purge
	
	local reactiveTaskDelays = {
		progressiveBackoff = false,
		errorDelay = 5,
		maxDelay = 60,
	}
	
	local reactiveTask = function()
	
		if self._debugProactivePurge then
			return Observable.never()
		end
		
		return ongoingReactivePurge( self )
		
	end

	self._reactivePurgeTask = WFRxContinuousTask( "NmbVolumeAgent:Reactive", reactiveTask, reactiveTaskDelays )

	-- Proactive Purge

	local proactiveTaskDelays = {
		progressiveBackoff = false,
		errorDelay = 5,
		maxDelay = 60,
	}
	 
	local proactiveTask = function() 
	
		if optOverrides.disableProactivePurge or self._debugReactivePurge then
		
			-- SPAM [jandrew 2019-10-29]: For external volumes, we disable proactive purge and
			-- entirely rely on reactive purge. Proactive purge is more useful on the main volume,
			-- and it does add extra processing. Disabling it on external volumes may help
			-- with some of the NAS performance issues that we've seen.
			
			return Observable.never()
			
		end
	
		return ongoingProactivePurge( self ) 
		
	end

	self._proactivePurgeTask = WFRxContinuousTask( "NmbVolumeAgent:Proactive", proactiveTask, proactiveTaskDelays )

	WFDiagnosticLog.addLogProvider( self )

end

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

function NmbVolumeAgent:close()

	check( 'NmbVolumeAgent' )
	
	local reactivePurgeTask = self._reactivePurgeTask
	self._reactivePurgeTask = nil
	
	if reactivePurgeTask then
		reactivePurgeTask:close()
	end

	local proactivePurgeTask = self._proactivePurgeTask
	self._proactivePurgeTask = nil
	 
	if proactivePurgeTask then
		proactivePurgeTask:close()
	end

	local purgeOnDelete = self._purgeOnDelete
	self._purgeOnDelete = nil
	
	if purgeOnDelete then
		purgeOnDelete:dispose()
	end

	local volumeStatusStream = self._volumeStatusStream
	self._volumeStatusStream = nil
	
	if volumeStatusStream then
		volumeStatusStream:dispose()
	end

	local volumeConnectionSubscription = self._volumeConnectionSubscription
	self._volumeConnectionSubscription = nil
	
	if volumeConnectionSubscription then
		volumeConnectionSubscription:dispose()
	end

	local reactivePurgeRunComplete = self._reactivePurgeRunComplete
	self._reactivePurgeRunComplete = nil
	
	if reactivePurgeRunComplete then
		reactivePurgeRunComplete:dispose()
	end

	local proactivePurgeRunComplete = self._proactivePurgeRunComplete
	self._proactivePurgeRunComplete = nil
	
	if proactivePurgeRunComplete then
		proactivePurgeRunComplete:dispose()
	end

end

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

function NmbVolumeAgent:stop()

	check( 'NmbVolumeAgent' )

	self._reactivePurgeTask:stopContinuousTask()
	self._proactivePurgeTask:stopContinuousTask()
	
end

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

function NmbVolumeAgent:start()

	check( 'NmbVolumeAgent' )

	self._reactivePurgeTask:startContinuousTask()
	self._proactivePurgeTask:startContinuousTask()
	
end

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

--[[!

	\brief Ask the volume agent to provide n bytes of free space.
	
	\param spaceNeeded (number) The number of bytes needed.
 	\param forcePurge (boolean) If true, this API will purge assets to make
		space, even if the resulting amount of free space is less than
		spaceNeeded. If false or nil, this API will _only_ perform any
		purging in the event that the amount of space able to reclaimed
		is the equal to or greater than spaceNeeded.
	
	\return (number) The amount of free space, in bytes, that we can provide.

--]]

function NmbVolumeAgent:tryToClaimSpace( spaceNeeded, forcePurge )

	check( 'NmbVolumeAgent', 'number', '?boolean' )

	if not self._isOnline then
		return Observable.just( 0 )
	end
	
	local startTime = KSDate.timeNowForAnimation()
	
	return getDiskSpaceNeededResultStream( self, spaceNeeded )
	
		:flatMapLatest( function( spaceNeededToPreventNoDiskSpace )
		
			if spaceNeededToPreventNoDiskSpace == 0 then
			
				return Observable.just( spaceNeeded )
					-- If we don't need to reclaim any space, we're done.
					
			end
		
			-- Attempt to reclaim the needed space.
		
			return self._reactivePurgeInProgressStream
	
				:filter( function( reactivePurgeInProgress )
				
					if not self._reactivePurgeInProgress then
					
						return true
					
					end
					
					logInfof( self, "Reactive purge already in progress. Waiting..." )	
					
					return false
					
				end )
				
				:take( 1 )
				
				:flatMapLatest( function()
				
					local maybePurgeReason = not forcePurge and reactivePurgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING
				
					return reclaimSpaceAndReport( self, maybePurgeReason, spaceNeededToPreventNoDiskSpace, 0, startTime )
					
				end )
				
				:map( function( spaceNeededAfterPurgeToPreventNoDiskSpace )
				
					return spaceNeeded - math.max( 0, spaceNeededAfterPurgeToPreventNoDiskSpace )
					
				end )
		
		end )
		
end

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

--[[!

	\brief Protect a list of assets from being purged. This API can be used
	during operations such as export to prevent certain needed assets from
	being removed from the system.
	
	\param assetIds (table) A table containing asset ids or local ids.
 
	\return (void)

--]]

function NmbVolumeAgent:protectFromPurge( assetIds )

	check( 'NmbVolumeAgent', 'table' )

	for _, assetId in ipairs( assetIds ) do
	
		local protectedCount = self._protectedFromPurge[ assetId ] or 0
		self._protectedFromPurge[ assetId ] = protectedCount + 1
		
	end

end

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

--[[!

	\brief Releases a list of assets from protection from the volume agent. Once,
	released, the assets are subject to standard purging.
	
	\param assetIds (table) A table containing asset ids or local ids.
 
	\return (void)

--]]

function NmbVolumeAgent:releaseProtectionFromPurge( assetIds )

	check( 'NmbVolumeAgent', 'table' )

	for _, assetId in ipairs( assetIds ) do
	
		local protectedCount = self._protectedFromPurge[ assetId ] or 0
		local newProtectedCount = protectedCount - 1
		
		self._protectedFromPurge[ assetId ] = newProtectedCount > 0 and newProtectedCount or nil
		
	end

end

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

--[[!

	\brief Returns a stream that fires periodically with the status of the disk.
	\return (Rx ReplaySubject)

--]]

function NmbVolumeAgent:getVolumeStatusStream()

	check( 'NmbVolumeAgent' )
	return self._volumeStatusStream

end

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

--[[!

	\brief Returns the volume GUID that the agent is operating on.
	\return (string) volume guid

--]]

function NmbVolumeAgent:getVolumeGuid()

	check( 'NmbVolumeAgent' )
	return self._volumeGuid

end

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

--[[!

	\brief Returns a stream that fires when global sync is paused or resumed.
	\return (Rx Subject)

--]]

function NmbVolumeAgent:isSyncPausedStream()

	check( 'NmbVolumeAgent' )
	return self._isSyncPausedStream

end

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

function NmbVolumeAgent:setSyncPausedStatus( isSyncPaused )

	check( 'NmbVolumeAgent', 'boolean' )
	self._isSyncPausedStream:onNext( isSyncPaused )
	
end

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

--[[!

	\brief Returns a stream that fires when the internet goes offline or online.
	\return (Rx Subject)

--]]

function NmbVolumeAgent:hasInternetConnectionStream()

	check( 'NmbVolumeAgent' )
	return self._hasInternetConnectionStream

end

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

function NmbVolumeAgent:setInternetConnectionStatus( hasInternetConnection )

	check( 'NmbVolumeAgent', 'boolean' )
	self._hasInternetConnectionStream:onNext( hasInternetConnection )

end

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

--[[!

	\brief Returns a stream that fires each time reactive purge runs to completion.
	
	THIS IS ONLY USED IN UNIT TESTS
	
	\return (Rx Subject)

--]]

function NmbVolumeAgent:reactivePurgeRunCompleteStream()

	check( 'NmbVolumeAgent' )
	return self._reactivePurgeRunComplete

end

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

--[[!

	\brief Returns a stream that fires each time proactive purge runs to completion.
	
	THIS IS ONLY USED IN UNIT TESTS
	
	\return (Rx Subject)

--]]

function NmbVolumeAgent:proactivePurgeRunCompleteStream()

	check( 'NmbVolumeAgent' )
	return self._proactivePurgeRunComplete

end

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

function NmbVolumeAgent:sendDiagnosticLogData()

	local WFFolderWalker = require 'WFFolderWalker'

	local uniqueId = KSIdGenerator.uuidInLowercaseHex()

	local function justReturn( ... )
		return ...
	end

	local function bytesToGBStr( bytes )
	
		if type( bytes ) == 'number' then
			return string.format( "%.5f GB", bytesToGb( bytes ) )
		else 
			return bytes
		end
		
	end

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

	local nimbusPath = buildAbsoluteNimbusPath( self )

	local sizeOfProxiesFolder
	local sizeOfOriginalsFolder
	local sizeOfCatalogFolder
	local numberOfProxiesInProxiesFolder
	local numberOfOriginalsInOriginalsFolder
	
	if nimbusPath then

		sizeOfProxiesFolder = getSizeOfNegativesResultStream( self, nimbusPath )
		:take( 1 )
		:map( function( _, _, sizeOfProxiesFolder )
			return bytesToGBStr( sizeOfProxiesFolder )
		end )
		
		sizeOfOriginalsFolder = getSizeOfNegativesResultStream( self, nimbusPath )
		:take( 1 )
		:map( function( _, sizeOfOriginalsFolder )
			return bytesToGBStr( sizeOfOriginalsFolder )
		end )
		
		sizeOfCatalogFolder = WFAsyncFileUtils.calculatePathSize( nimbusPath )
		:map( function( size )
		
			return bytesToGBStr( size )
		
		end )
		
		local function countFilesInDirectory( dir )
	
			return Observable.just( true )
		
			:map( function()
			
				local count = 0
				
				for _ in WFFolderWalker( dir ) do
					count = count + 1
				end
				
				return count
				
			end )
		
		end

		numberOfProxiesInProxiesFolder
			= countFilesInDirectory( child( nimbusPath, 'proxies' ) )
			
		numberOfOriginalsInOriginalsFolder
			= countFilesInDirectory( child( nimbusPath, 'originals' ) )

	end

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

	local indexStoreNumberOfLocalOriginals = self._indexStore:queryOnce{
			sql = [[ SELECT COUNT(*) 
						FROM assetPathsAndInfo 
						WHERE originalVolumeGuid IS NOT NULL ]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true
		}
		:map( function( results )
			return results[ 1 ]
		end )

	local indexStoreNumberOfLocalProxies = self._indexStore:queryOnce{
			sql = [[ SELECT COUNT(*) 
						FROM assetPathsAndInfo 
						WHERE proxyVolumeGuid IS NOT NULL ]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true
		}
		:map( function( results )
			return results[ 1 ]
		end )

	local indexStoreSizeOfLocalOriginals = self._indexStore:queryOnce{
			sql = [[ SELECT SUM( originalSize ) 
						FROM assetPathsAndInfo 
						WHERE originalVolumeGuid IS NOT NULL ]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true
		}
		:map( function( results )
			return bytesToGBStr( results[ 1 ] )
		end )

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

	local mruNumberOfLocalOriginals = self._fileAccessDb:_DEBUG_numberOfLocalOriginals()

	local mruSizeOfLocalOriginals = self._fileAccessDb:_DEBUG_sizeOfLocalOriginals()
		:map( function( size )
			return bytesToGBStr( size )
		end )
		
	local mruNumberOfLocalProxies = self._fileAccessDb:_DEBUG_numberOfLocalProxies()

	local mruSizeOfLocalProxies = self._fileAccessDb:_DEBUG_sizeOfLocalProxies()
		:map( function( size )
			return bytesToGBStr( size )
		end )
		
	----------------------------------------------------------------------------

	return {

		name = "Volume Agent:" .. uniqueId,
		id = 'NmbVolumeAgent:' .. uniqueId,
		weight = 510,
		parentSection = WFDiagnosticLog.Sections.LOCAL_STORAGE,
		
		body = {
		
			-- The following metrics are gathered by querying the OS directly. This should
			-- be the reality of what's on disk.
			
			[ 'HD: Free Space on Disk' ] = bytesToGBStr( self._remainingSpace ),
			
			[ 'HD: Size of Catalog Folder' ] = sizeOfCatalogFolder,
			
			[ 'HD: Number of Local Originals in Originals Folder' ] = numberOfOriginalsInOriginalsFolder,
			[ 'HD: Size of Local Originals Folder' ] = sizeOfOriginalsFolder,

			[ 'HD: Number of Local Proxies in Proxies Folder' ] = numberOfProxiesInProxiesFolder,
			[ 'HD: Size of Local Proxies Folder' ] = sizeOfProxiesFolder,
	
			-- The following metrics are gathered from the docstore. Discrepencies between the
			-- number of binaries and the size of binaries between this section and the
			-- MRU.db / HD sections could be indicative of a bug or file leak.
			
	
			[ 'Index Store: Number of Local Originals' ] = indexStoreNumberOfLocalOriginals,
			[ 'Index Store: Size of Local Originals' ] = indexStoreSizeOfLocalOriginals,
			[ 'Index Store: Number of Local Proxies' ] = indexStoreNumberOfLocalProxies,
			
			-- The following metrics are gathered from the MRU.db, the database used
			-- by the volume agent to drive purge. Discrepencies between the
			-- number of binaries and the size of binaries between this section and the
			-- HD / Index Store sections could be indicative of a bug or file leak.
			

			[ 'MRU.db: Volume GUID' ] = self._volumeGuid,

			[ 'MRU.db: Is storing all originals locally?' ] = self._keepAllOriginals,
			[ 'MRU.db: Is storing all proxies locally?' ] = self._keepAllProxies,
			
			[ 'MRU.db: Target LrD Disk Space Usage Percentage' ] = self._targetDiskSpaceUsage,
		
			[ 'MRU.db: Number of Local Originals' ] = mruNumberOfLocalOriginals,
			[ 'MRU.db: Size of Local Originals' ] =  mruSizeOfLocalOriginals,
			
			[ 'MRU.db: Number of Local Proxies' ] = mruNumberOfLocalProxies,
			[ 'MRU.db: Size of Local Proxies' ] = mruSizeOfLocalProxies,
			
			[ 'MRU.db: Size of Purgeable Proxies' ] = bytesToGBStr( self._maxSizeOfPurgeableProxies ),
			
			[ 'MRU.db: Is LrD Over Target Disk Space Usage' ] = self._isOverTargetDiskSpaceUsage,
			[ 'MRU.db: Is LrD Out of Idle Space' ] = self._isOutOfIdleSpace,
			[ 'MRU.db: Is HD Out of Space' ] = self._isOutOfSpace,
			
			[ 'MRU.db: The Purge Log' ] = WFRingBuffer.map( justReturn, self._debugLogRingBuffer ),

		}

	}

end

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

return Rx.Class 'NmbVolumeAgent'( NmbVolumeAgent )
