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

WFCatVolumeAgent.lua

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

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

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

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

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

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

	-- WF.oz_client.managed_catalog
local WFCatLocalStorage = require 'WFCatLocalStorage'
local WFCatPurgeHelper = require 'WFCatPurgeHelper'
local WFCatPurgeOnDelete = require 'WFCatPurgeOnDelete'
local WFCatVolumeAgentQueryHelper = require 'WFCatVolumeAgentQueryHelper'

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

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

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

local NmbFileManagementSettings
local LrmVolumeAgentSettings, LrmOzPurgeAgentUtils, LrmConfig

if WFPlatformInfo.isDesktop then
	NmbFileManagementSettings = require 'NmbFileManagementSettings'
else
	LrmVolumeAgentSettings = require 'LrmVolumeAgentSettings'
	LrmOzPurgeAgentUtils = require 'LrmOzPurgeAgentUtils'
	LrmConfig = require 'LrmConfig'
end

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

local DISABLE_PURGE = WFConfiguration.WFCatVolumeAgent.disablePurge

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

local check = assert( WFArguments.check )

local child = assert( KSPathUtils.child )

local runUntilDone = assert( WFRxRunUntilDone )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

local KSTableUtils_setNested = assert( KSTableUtils.setNested )

local storageByteString = assert( KSStringUtils.storageByteString )

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

local kDiskSpaceCheckPeriod = 30
	-- Even if there are no changes within the app, we run a disk space check
	-- periodically to see if volume space is low. This value represents how 
	-- frequently we run this check.

local kBufferPercentage = 0.25
	-- The percentage of space within the cache that is attempted to be reclaimed
	-- by reactive purge.

local kMaxDebugLogSize = 100
	-- The maximum size of the debug log that is included in the diagnostic log.
	
local kMaxSizeOfProtectedRingBuffer = 10
	-- The maximum size of the protected ring buffer.
	
local kProactivePurgeFrequencyInDays = WFPlatformInfo.isMobile() and 7 or 1
	-- The frequency at which proactive purge can run.
	-- WF-1938: Proactive Purge on Mobile is intentionally less aggressive.
	
--------------------------------------------------------------------------------

local purgeReasons = {

	NO_DISK_SPACE 					= 1, 
		-- Triggers when the drive is out of disk space.
	TARGET_RANGE 					= 2, 
		-- Triggers when the cache grows beyond the user-specified size.
	NO_DISK_SPACE_ALL_OR_NOTHING	= 3,
		-- Triggered by the tryToClaimSpace API.
	PROACTIVE_PROXIES				= 4,
		-- LrD Only: Proactive purge of proxies. Triggers on startup and every 24 hours.
	PROACTIVE_ORIGINALS				= 5,
		-- LrD Only: Proactive purge of originals. Triggers on startup and every 24 hours.
	PROACTIVE_ALL					= 6,
		-- Mobile Only: Proactive purge of everything. Triggers on startup and every 24 hours.
	EMPTY_CACHE						= 7
		-- Triggered by the clearCache API, used to empty the entire cache.
}	

local purgeReasonStrings = KSTableUtils.invert( purgeReasons )

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

-- @mark Analytics/Logging

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

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

	-- Adds a log entry to the ring buffer, and emits a logger statement.

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

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

end

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

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

	-- Adds a logger warning entry to the ring buffer, and emits a warning statement.

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

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

end

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

local function logReactivePurgeInfo( failed, desc, currentCacheSize, appAndUserDataSize,
		locallyStoredSize, currentFreeSpace )

	local settings = WFPlatformInfo.isDesktop and NmbFileManagementSettings or LrmVolumeAgentSettings

	local event = {}

	event[ 'event.type' ] = failed and 'error' or 'info'
	event[ 'event.description' ] = failed and 'Reactive Purge Failed' or 'Reactive Purge Succeeded'
	event[ 'event.appWorkflow' ] = 'PURGE'
	event[ 'event.subcategory' ] = 'Reactive'
	event[ 'event.purgeTrigger' ] = desc
	event[ 'event.targetCacheSize' ] = storageByteString( settings.targetCacheSize() )
	event[ 'event.currentCacheSize' ] = currentCacheSize and storageByteString( currentCacheSize )
	event[ 'event.minimumFreeSpace' ] = storageByteString( settings.minimumFreeSpace() )
	event[ 'event.currentFreeSpace' ] = currentFreeSpace and storageByteString( currentFreeSpace )
	event[ 'event.appAndUserDataSize' ] = appAndUserDataSize and storageByteString( appAndUserDataSize )
	event[ 'event.locallyStoredSize' ] = locallyStoredSize and storageByteString( locallyStoredSize )
	
	WFAnalytics.trackEvent( event )
	
end

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

local function logProactivePurgeInfo( self, spaceReclaimed, reason, currentCacheSize )

	local settings =  WFPlatformInfo.isDesktop and NmbFileManagementSettings or LrmVolumeAgentSettings

	local event = {}

	local binaryTypePurged
	
	if reason == purgeReasons.PROACTIVE_PROXIES then
		binaryTypePurged = 'Proxies'
	elseif reason == purgeReasons.PROACTIVE_ORIGINALS then
		binaryTypePurged = 'Originals'
	elseif reason == purgeReasons.PROACTIVE_ALL then
		binaryTypePurged = 'All'
	else
		binaryTypePurged = 'Unknown'
	end

	event[ 'event.type' ] = 'info'
	event[ 'event.appWorkflow' ] = 'PURGE'
	event[ 'event.subcategory' ] = 'Proactive'
	event[ 'event.binaryTypePurged' ] = binaryTypePurged
	event[ 'event.targetCacheSize' ] = storageByteString( settings.targetCacheSize() )
	event[ 'event.currentCacheSize' ] = currentCacheSize and storageByteString( currentCacheSize )
	event[ 'event.spaceReclaimed' ] = spaceReclaimed and storageByteString( spaceReclaimed )
	
	WFAnalytics.trackEvent( event )
	
	logInfof( self, "Proactive Purge of %s Binaries Complete. Results:", binaryTypePurged )
	logInfof( self, "Max Photo Cache Size %s, Cache Size After Proactive Purge %s, Space Reclaimed %s",
			storageByteString( settings.targetCacheSize() ), storageByteString( currentCacheSize ),
			storageByteString( spaceReclaimed ) )

end

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

local function logLocalStorageOptions( settings )

	local event = {}

	event[ 'event.type' ] = 'info'
	event[ 'event.appWorkflow' ] = 'PURGE'
	event[ 'event.subcategory' ] = 'Settings'
	event[ 'event.targetCacheSize' ] = storageByteString( settings.targetCacheSize() )
	
	if WFPlatformInfo.isDesktop then
	
		local originalStorageLocationInfo = settings.originalStorageLocationInfo()
	
		event[ 'event.keepAllOriginals' ] = settings.keepAllOriginals()
		event[ 'event.keepAllProxies' ] = settings.keepAllProxies()
		event[ 'event.originalStorageLocation' ] = originalStorageLocationInfo.readableIdentifier and 'custom' or 'default'
		
	end
	
	WFAnalytics.trackEvent( event )
	
end

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

local function logDiskStats( freeSpace, totalSpace, sizeOfCache )

	local event = {}

	event[ 'event.type' ] = 'info'
	event[ 'event.appWorkflow' ] = 'PURGE'
	event[ 'event.subcategory' ] = 'DiskStats'
	event[ 'event.freeSpace' ] = freeSpace and storageByteString( freeSpace )
	event[ 'event.totalDeviceStorage' ] = totalSpace and storageByteString( totalSpace )
	event[ 'event.sizeOfCache' ] = sizeOfCache and storageByteString( sizeOfCache )
	
	WFAnalytics.trackEvent( event )
	
end

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

local function maybeLogDiskStats( self, freeSpace, totalSpace, sizeOfCache )
	
	if not self._hasSentDiskStats
			and freeSpace
			and totalSpace
			and sizeOfCache then
	
		self._hasSentDiskStats = true
  
  		local settings =  WFPlatformInfo.isDesktop and NmbFileManagementSettings or LrmVolumeAgentSettings

		logLocalStorageOptions( settings )
		
		logDiskStats( freeSpace, totalSpace, sizeOfCache )

	end

end

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

local getSizeOfLrCache, getFreeSpace, isProactivePurgeReason -- Forward declarations

local function maybeLogPurgeInfo( self, purgeReason, desc,
		spaceNeededAfterPurgeToPreventNoDiskSpace, spaceNeededAfterPurgeToReachTargetDiskSpace )

	-- Proactive purge cannot fail, as there's no right or wrong amount of bytes
	-- to purge.
	
	if not isProactivePurgeReason( purgeReason )
			and purgeReason ~= purgeReasons.EMPTY_CACHE then

		local didReactivePurgeFail
		
		if ( spaceNeededAfterPurgeToPreventNoDiskSpace > 0 
					and not self._isOutOfSpace ) or
				( spaceNeededAfterPurgeToReachTargetDiskSpace > 0 
						and not self._isOverTargetDiskSpaceUsage ) then

			-- This condition checks if purge failed. Additionally, it dedupes failure
			-- notifications to avoid spamming Dunamis.

			didReactivePurgeFail = true

		elseif spaceNeededAfterPurgeToPreventNoDiskSpace <= 0
				and spaceNeededAfterPurgeToReachTargetDiskSpace <= 0
				and WFPlatformInfo.isMobile() then
				
			-- This condition checks if it succeeded. We only log successes for mobile clients.
				
			didReactivePurgeFail = false
			
		end
		
		if didReactivePurgeFail ~= nil then

			return getSizeOfLrCache( self )
			
				:map( function( photoCacheSize, appAndUserDataSize, locallyStoredSize )
				
					local freeSpace = getFreeSpace( self )
				
					freeSpace = math.max( 0, freeSpace - self._minimumFreeSpace )
				
					logReactivePurgeInfo(
							didReactivePurgeFail, desc, photoCacheSize,
							appAndUserDataSize, locallyStoredSize, freeSpace )
					
					return spaceNeededAfterPurgeToPreventNoDiskSpace,
							spaceNeededAfterPurgeToReachTargetDiskSpace
				
				end )
		
		end
	
	end
	
	return Observable.just(
			spaceNeededAfterPurgeToPreventNoDiskSpace,
			spaceNeededAfterPurgeToReachTargetDiskSpace )

end

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

local function logProactivePurgeResults( self, spaceReclaimed, reason )

	return getSizeOfLrCache( self )
	
		:tap( function( photoCacheSize )

			logProactivePurgeInfo( self, spaceReclaimed, reason, photoCacheSize )
		
		end )

end

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

-- @mark Misc

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

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

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

local function isAllOrNothingReason( reason )

	return reason == purgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING
		
end

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

isProactivePurgeReason = function( reason )

	return reason == purgeReasons.PROACTIVE_PROXIES
			or reason == purgeReasons.PROACTIVE_ORIGINALS
			or reason == purgeReasons.PROACTIVE_ALL
		
end

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

local function getVolumeFromGuid( self )

	local volume

	if self._volumeGuid then
		volume = WFFileUtils.getVolumeFromGuid( self._volumeGuid )
	else
		volume = '/'
	end

	if not volume then
		logInfof( self, "Failed to get volume name from volume GUID %s", self._volumeGuid )
	end
	
	return volume

end

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

local function updateVolumeStatusStream( self )

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

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

local function checkAvailableSpaceOnVolume( _, volume )

	-- Calculate how much free space is on the system, excluding the no-disk-space
	-- buffer. This function is not actually used by the Volume Agent, but is used by
	-- other LrD tasks.

	check( 'table', 'string' )

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

	if not success then
		return 0
	end
			
	local reservedSpace = WFPlatformInfo.isMobile()
			and LrmVolumeAgentSettings.minimumFreeSpace()
			or NmbFileManagementSettings.minimumFreeSpace()
			
	return math.max( 0, freeSpaceOnVolume - reservedSpace )

end

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

local function isDirectoryReadOnly( path )

-- Return true if this directory is read only.
-- if the directory doesn't exist check up the path until we find one that
-- that exists and return the status of that as any directories we create
-- within should inherit the read/write permissions from above.

	local prevPath = nil
	while path ~= "" and path ~= prevPath and not WFFileUtils.exists( path ) do
		prevPath = path
		path = KSPathUtils.parent( path )
	end

	return path and path ~= "" and WFFileUtils.isDirectoryReadOnly( path )

end

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

local function isVolumeOnlineStream( self )

	-- Returns a stream that fires periodically with the status of the volume.
	-- The stream fires with two values:
	
	-- isOnline: true if the volume is connected to the device, false otherwise.
	-- isReadOnly: false if the volume is readable and writeable, true if
	-- volume is readOnly or path name of read only directory encountered.

	local maybeTimer = Observable.just( true )
	
	if self._volumeStatusCheckPeriod > 0 then
		maybeTimer = Observable.timer( 0, self._volumeStatusCheckPeriod, WFRxScheduler.sharedScheduler() )
	end

	return maybeTimer
	
		:map( function()
		
			local volume = getVolumeFromGuid( self )
		
			local isOnline = not not volume
				-- If we can extract the volume name from the volume guid, the volume is online.

			local isReadOnly = false
			
			if WFPlatformInfo.isDesktop and isOnline then

				if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then
					return isOnline, false
				end

				local originalsPathInfo = self._catalog:fileSpec():originalsPathInfo() or {}

				-- If the originals path isn't stored in the normal location
				-- the user may have modified permissions, verify they are not readonly.

				if self._volumeGuid == originalsPathInfo.volumeGuid then

					local success, originalsRootPath = pcall(
							WFFileUtils.createFullPathFromRelativePathAndGuid,
									originalsPathInfo.relativePath, originalsPathInfo.volumeGuid )

					if success and type( originalsRootPath ) == 'string' then

						local originalsFolder = child( originalsRootPath, 'originals' )

						-- Return the originals folder so we can provide better messaging.
						isReadOnly = isDirectoryReadOnly( originalsFolder ) and originalsFolder

					end

				else

					isReadOnly = WFFileUtils.isVolumeReadOnly( self._volumeGuid )

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

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

local function getUntrackedBinariesRemover( self )

	return WFCatVolumeAgentQueryHelper.untrackedBinariesPurgeListStream( self._catalog )

		:filter( function( purgeList )
		
			return next( purgeList )
		
		end )
		
		:flatMap( function( purgeList )

			if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then
				return Observable.just( false )
			end

			return WFCatPurgeHelper.deleteFilesByPurgeList(
					self._catalog, purgeList, self._idleDownloader, self._dataCache )
		
		end)

end

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

-- @mark Purge Algorithm

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

local function isProtectedFromPurge( self, protectedFromPurge, localId, docId )

	-- Returns true if the specified localId or docId is protected from purge.
	-- Returns false otherwise.

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

	return false

end

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

local function processBatchOfBinariesForPurge( self, binaryType, spaceNeeded,
												batchOfBinaries, getNextBatch,
												purgeList, purgeStream )

	-- This function processes a batch of purge candidates and remove entries
	-- that have been protected from purge (via the protectFromPurge API) as
	-- well as entries that have already been marked for purge.

	check( 'WFCatVolumeAgent', 'string', 'number', 'table', 'function', 'table', 'table' )

	if #batchOfBinaries == 0 then
	
		logInfof( self, "No more purge candidates to process." )
		purgeStream:onNext( purgeList, spaceNeeded )
		purgeStream:onCompleted()
		return
		
	end
	
	logInfof( self, "Now processing %d %s purge candidates.", #batchOfBinaries, binaryType )

	for _, binaryInfo in ipairs( batchOfBinaries ) do
	
		if spaceNeeded <= 0 then
			break
		end
	
		local localId = binaryInfo[ 1 ]
		local docId = binaryInfo[ 2 ]
		local fileSize = binaryInfo[ 3 ] or 0
		local maybePurgeSpecifics = binaryInfo[ 4 ] or true
		
		if type( maybePurgeSpecifics ) == 'table' 
				and not next( maybePurgeSpecifics ) then
		
			maybePurgeSpecifics = true
		
		end

		-- If the document is not protected from purge, and it hasn't been
		-- previously marked for purge, add it to the purge list.

		if not isProtectedFromPurge( self, self._protectedFromPurge, localId, docId ) then

			local isMarkedForPurge = KSTableUtils_derefOpt(
					self._isBinaryMarkedForPurgeByTypeAndLocalId, binaryType, localId )
			
			if not isMarkedForPurge then

				KSTableUtils_setNested(
						self._isBinaryMarkedForPurgeByTypeAndLocalId, true, binaryType, localId )
					-- This is used to prevent purging the same binary more than once.

				local purgeListForBinaryType = purgeList[ binaryType ] or {}
				purgeListForBinaryType[ localId ] = maybePurgeSpecifics
				purgeList[ binaryType ] = purgeListForBinaryType
				
				spaceNeeded = spaceNeeded - fileSize

			end
		
		end
	
	end
		
	if spaceNeeded > 0 then
	
		-- Get the next batch of binaries and continue purging.
		
		Observable.defer( function()
		
			if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then

				purgeStream:onNext( {}, spaceNeeded )
				purgeStream:onCompleted()

				return Observable.just( true )

			end

			return getNextBatch()

				:map( function( newBatchOfBinaries, newGetNextBatch )
			
					processBatchOfBinariesForPurge( self, binaryType, spaceNeeded,
							newBatchOfBinaries, newGetNextBatch, purgeList, purgeStream )
			
					return newBatchOfBinaries, newGetNextBatch
			
				end )
				
		end )
		
		:call( runUntilDone )
		
	else
	
		-- We've identified enough binaries to purge, so we're done!
	
		purgeStream:onNext( purgeList, spaceNeeded )
		purgeStream:onCompleted()
	
	end
		
end

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

local function fetchBinariesToPurge( self, isProactivePurge, binaryType, spaceNeeded, purgeList )

	-- This function fetches the first batch of purge candidates and starts
	-- processing them. See the API definitions within WFCatVolumeAgentQueryHelper
	-- for more information.

	check( 'WFCatVolumeAgent', 'boolean', 'string', 'number', 'table' )

	local getBatchOfBinariesFn = WFCatVolumeAgentQueryHelper.getBatchOfPurgeableBinaries

	if isProactivePurge then
	
		-- If purging proactively, we only consider inactive (AKA unimportant) binaries
		-- for purge. Proactive purge does not remove binaries that the system believes
		-- are still useful to the user.
	
		getBatchOfBinariesFn = WFCatVolumeAgentQueryHelper.getBatchOfInactiveBinaries
		
	end

	local processBinariesStream = Rx.ReplaySubject( 1 )
	
	if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then

		processBinariesStream:onNext( {}, spaceNeeded )
		processBinariesStream:onCompleted()

	end

	getBatchOfBinariesFn( self._catalog, binaryType, self._volumeGuid )
	
		:map( function( batchOfBinaries, getNextBatch )
	
			processBatchOfBinariesForPurge( self, binaryType, spaceNeeded,
					batchOfBinaries, getNextBatch, purgeList, processBinariesStream )
		
			return batchOfBinaries, getNextBatch
		
		end )
		
		:call( runUntilDone )
		
	return processBinariesStream

end

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

local function restructurePurgeList( self, purgeList )

	-- Turns the purge list into a format understood by WFCatPurgeHelper.lua

	local newPurgeList = {}

	for binaryType, values in pairs( purgeList ) do
	
		values.binaryType = binaryType
		newPurgeList[ #newPurgeList + 1 ] = values

	end
	
	return newPurgeList

end

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

local function purgeToReclaimSpace( self, purgeReason, binaryTypesToPurge, spaceNeeded,
									binaryTypesIndex, purgeResultStream, purgeList )

	-- The basic algorithm used by purgeToReclaimSpace is as follows:
	-- 1. Start with the first binary type within the binaryTypesToPurge table.
	-- 2. Query for purge candidates matching this specified type.
	-- 3. If the file size of all the purge candidates is greater than or equal to
	-- 		the amount of space we need to reclaim, then we're done. Otherwise, go to
	-- 		step 2 using the next binaryType within binaryTypesToPurge.

	check( 'WFCatVolumeAgent', 'number', 'table', 'number', '?number', '?table', '?table' )

	-- Figure out which binary type to purge
	
	binaryTypesIndex = binaryTypesIndex or 1
	local binaryType = binaryTypesToPurge[ binaryTypesIndex ]

	-- Maybe create a purge result stream which is used to signal to the caller when
	-- 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 binaries all at once (which
	-- is a lot more performant that purging binaries one-at-a-time).

	purgeList = purgeList or {}
	
	logInfof( self, "Attempting to reclaim %s of disk space using %s",
		storageByteString( spaceNeeded ), binaryType )
		
	local isProactivePurge = isProactivePurgeReason( purgeReason )
		
	fetchBinariesToPurge( self, isProactivePurge, binaryType, spaceNeeded, purgeList )
		
		:flatMapLatest( function( purgeList, spaceNeeded )
		
			if spaceNeeded > 0 then

				logInfof( self, "Still need to reclaim %s of disk space.", storageByteString( spaceNeeded ) )
	
				binaryTypesIndex = binaryTypesIndex + 1
				
				if binaryTypesToPurge[ binaryTypesIndex ] then
				
					-- We need to reclaim more space so attempt to purge the next binary type
					
					purgeToReclaimSpace( self, purgeReason, binaryTypesToPurge, spaceNeeded,
											binaryTypesIndex, purgeResultStream, purgeList )
							
					return Observable.empty()
					
				end
				
			end

			local newPurgeList = restructurePurgeList( self, purgeList )

			local maybeDeleteAssets = Observable.just( true )			

			local allOrNothing = isAllOrNothingReason( purgeReason )

			local didDeleteFiles = false

			-- If allOrNothing is true, it means that we should only perform the purge
			-- if we're able to reclaim *all* of the space specified by spaceNeeded. This
			-- is only true if purge is initiated by the tryToClaimSpace API.

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

				didDeleteFiles = true

				logInfof( self, "Now Deleting Files & Updating Document Annotations" )

				maybeDeleteAssets = WFCatPurgeHelper.deleteFilesByPurgeList(
						self._catalog, newPurgeList, self._idleDownloader, self._dataCache )

			else
			
				logInfof( self, "Abandoning Purge. Nothing will be purged." )
			
			end
			
			local function completePurgeResultStream()
			
				self._debugPurgeListStream:onNext( purgeList )
			
				purgeResultStream:onNext( spaceNeeded, purgeList, didDeleteFiles )
				purgeResultStream:onCompleted()
				
			end
			
			return maybeDeleteAssets:finally( completePurgeResultStream )
			
		end )
		
		:call( runUntilDone )

	return purgeResultStream

end

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

local function getBinaryTypesToPurge( maybePurgeReason, keepAllOriginals, keepAllProxies )
	
	-- Build up binaryTypesToPurge, which lists what binary types are purgeable
	-- for each platform.
	
	-- The order in which the binaries are added into binaryTypesToPurge is the
	-- order that the binaries are purged.

	local isProactivePurge = isProactivePurgeReason( maybePurgeReason )

	local binaryTypesToPurge = {}

	if WFPlatformInfo.isMobile() then
	
		if not isProactivePurge then
	
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'original'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'proxy'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'xmp_develop'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'aux'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'fullsize'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'preview'

		else

			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'original'

		end

	else
	
		if ( not isProactivePurge or maybePurgeReason == purgeReasons.PROACTIVE_ORIGINALS  )
				and not keepAllOriginals then
				
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'original'
			
		end
		
		if ( not isProactivePurge or maybePurgeReason == purgeReasons.PROACTIVE_PROXIES  )
				and not keepAllProxies then
				
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'proxy'
			
		end
		
		if not isProactivePurge and not keepAllProxies and not keepAllOriginals then
		
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'xmp_develop'
			binaryTypesToPurge[ #binaryTypesToPurge + 1 ] = 'aux'
		
		end
		
	end
	
	return binaryTypesToPurge

end

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

local calculatePurgeBufferStream -- forward declaration

local function performPurge( self, maybePurgeReason, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace, desc )

	-- performPurge operates in two steps.
	
	-- The first step attempts to reclaim enough space so that the volume is
	-- not longer out of disk space. If the volume is not out of disk space when
	-- this function is called, step one is a noop.
	
	-- The second step differs depending on whether this function is called by
	-- reactive purge, or proactive purge. If called by reactive purge, step 2
	-- reclaims additional space so that the cache is correctly sized. If called
	-- by proactive purge, step 2 will try to remove as many inactive binaries
	-- from the cache as possible.

	-- FWIW this two-step architecture is cruft from the old NmbVolumeAgent, and
	-- should be simplified in the future.
	
	local binaryTypesToPurge = getBinaryTypesToPurge( maybePurgeReason, self._keepAllOriginals, self._keepAllProxies )
	
	if #binaryTypesToPurge == 0 then
		return Observable.just( spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace )
	end

	self._purgeInProgress = true
	self._purgeInProgressStream:onNext( self._purgeInProgress )

	if not maybePurgeReason and spaceNeededToPreventNoDiskSpace > 0 then
	
		logInfof( self, "Must purge %s of binaries to no longer be out of space.",
				storageByteString( spaceNeededToPreventNoDiskSpace ) )
				
	end
	
	if spaceNeededToReachTargetDiskSpace > 0 then
	
		if isProactivePurgeReason( maybePurgeReason ) then
		
			logInfof( self, "Will try to Proactive Purge %s of binaries.",
					storageByteString( spaceNeededToReachTargetDiskSpace ) )
					
		else
		
			logInfof( self, "Must purge %s of binaries to reach the maximum Photo Cache size.",
					storageByteString( spaceNeededToReachTargetDiskSpace ) )
					
		end
		
	end
	
	self._isBinaryMarkedForPurgeByTypeAndLocalId = {}
		-- Clear the cache of previously purged binaries
	
	local purgeReason = maybePurgeReason or purgeReasons.NO_DISK_SPACE
	
	local maybePurgeToReclaimSpace = Observable.just( 0 )
	
	if spaceNeededToPreventNoDiskSpace > 0 then

		-- Step 1: First purge assets to prevent a no-disk-space scenario.

		logInfof( self, "Attempting to purge %s of binaries to no longer be out of space.",
				storageByteString( spaceNeededToPreventNoDiskSpace ) )

		maybePurgeToReclaimSpace = purgeToReclaimSpace(
				self, purgeReason, binaryTypesToPurge, spaceNeededToPreventNoDiskSpace )

	end
	
	return maybePurgeToReclaimSpace

		:flatMapLatest( function( spaceNeededAfterPurgeToPreventNoDiskSpace )

			if not isProactivePurgeReason( purgeReason )
					and not isAllOrNothingReason( purgeReason ) then

				return calculatePurgeBufferStream( self, purgeReason )
					:map( function( purgeBuffer )
						return spaceNeededAfterPurgeToPreventNoDiskSpace, purgeBuffer
					end )

			end

			return Observable.just( spaceNeededAfterPurgeToPreventNoDiskSpace, 0 )

		end )

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

				spaceNeededToReachTargetDiskSpace = 
						spaceNeededToReachTargetDiskSpace + spaceNeededAfterPurgeToPreventNoDiskSpace

				-- Step 2:
				-- If Reactive Purge: Purge assets to reach the target disk space threshold.
				-- If Proactive Purge: Purge whatever inactive binaries can be purged.

				local spaceNeeded = spaceNeededToReachTargetDiskSpace + purgeBuffer

				if spaceNeeded < 0 then

					-- No need to purge anything else.

					return Observable.just(
							spaceNeededAfterPurgeToPreventNoDiskSpace,
							spaceNeededToReachTargetDiskSpace )

				end

				if isProactivePurgeReason( purgeReason ) then
					logInfof( self, "Attempting to Proactive Purge %s of binaries.",
							storageByteString( spaceNeeded ) )
				elseif spaceNeededToReachTargetDiskSpace > 0 then
					logInfof( self, "Attempting to purge %s of binaries to reach the target Photo Cache size.",
							storageByteString( spaceNeeded ) )
				else
					logInfof( self, "Attempting to purge an additional %s of binaries as a purge buffer.",
							storageByteString( spaceNeeded ) )
				end
				
				purgeReason = maybePurgeReason or purgeReasons.TARGET_RANGE
				
				return purgeToReclaimSpace( self, purgeReason, binaryTypesToPurge, spaceNeeded )
				
					:map( function( value )
						return value - purgeBuffer
					end )

					:map( function( spaceNeededAfterPurgeToReachTargetDiskSpace )
					
						return spaceNeededAfterPurgeToPreventNoDiskSpace,
								spaceNeededAfterPurgeToReachTargetDiskSpace
					
					end )
			
			end
		
			-- No-disk-space purge failed.
			
			return Observable.just(
					spaceNeededAfterPurgeToPreventNoDiskSpace,
					spaceNeededToReachTargetDiskSpace )
		
		end )
		
		:flatMap( function( spaceNeededAfterPurgeToPreventNoDiskSpace, spaceNeededAfterPurgeToReachTargetDiskSpace )
		
			return maybeLogPurgeInfo(
					self,
					purgeReason,
					desc,
					spaceNeededAfterPurgeToPreventNoDiskSpace,
					spaceNeededAfterPurgeToReachTargetDiskSpace )
		
		end )

		:finally( function()
		
			self._purgeInProgress = false
			self._purgeInProgressStream:onNext( self._purgeInProgress )
			
		end )
		
end

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

-- @mark Logic Determining HOW MUCH to Purge

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

getFreeSpace = function( self )

	if self._debugMockDiskFreeSpace then
	
		-- Used by unit tests
		
		local freeSpace = self._debugMockDiskFreeSpace
		self._remainingSpace = math.max( 0, freeSpace - self._minimumFreeSpace )
		updateVolumeStatusStream( self )
		
		return freeSpace
		
	end
	
	local path = self._volumeRootPath
	
	local success = true
	
	if self._volumeGuid then
	
		success, path = pcall( WFFileUtils.createFullPathFromRelativePathAndGuid, path, self._volumeGuid )
	
		if success then
		
			success = WFFileUtils.isVolumeGuidOnline( self._volumeGuid )
		
		end
	
	end
	
	local fileSystemFreeSpace, totalSpace
	
	if success then
	
		success, fileSystemFreeSpace, totalSpace = pcall( WFFileUtils.freeSpaceAtPath, path )

		fileSystemFreeSpace = success and fileSystemFreeSpace or nil
		
		if fileSystemFreeSpace then
			self._remainingSpace = math.max( fileSystemFreeSpace - self._minimumFreeSpace, 0 )
			updateVolumeStatusStream( self )
		end
		
	end
	
	return fileSystemFreeSpace, totalSpace

end

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

getSizeOfLrCache = function( self )

	if self._debugMockLrSize then
		return Observable.just( self._debugMockLrSize, 0, 0 )
	end
	
	return WFCatLocalStorage.getPhotoCacheSizeStream(
			self._catalog, self._volumeGuid, self._keepAllOriginals, self._keepAllProxies, self._binaryStrategy )

end

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

local function calculatePurgeBuffer( reason, currentCacheSize, targetCacheSize )

	-- Returns an observable that sends the size of the purgeBuffer and completes.
	-- The purgeBuffer is the amount of space that should we should attempt to
	-- reclaim in addition the the space we need to reclaim in order to reach
	-- the target cache size. For example, if the target cache size is 1.0GB,
	-- and the current cache size is 1.2 GB, we should reclaim 0.2GB of space
	-- plus the purge buffer so that cache is not completely full.
	
	local buffer = 0

	if reason == purgeReasons.PROACTIVE_ALL
			or reason == purgeReasons.PROACTIVE_ORIGINALS
			or reason == purgeReasons.PROACTIVE_PROXIES then
	
		-- Generally, for proactive purge, there's no limit to how much we can purge.
		
		buffer = currentCacheSize
		
	elseif currentCacheSize > targetCacheSize then

		buffer = targetCacheSize * kBufferPercentage

	elseif reason == purgeReasons.NO_DISK_SPACE then

		-- https://jira.corp.adobe.com/browse/WF-1921
		-- In the event that reactive purge is triggered due to low space on the device,
		-- and the photo cache is *not* too big, then we use a static purge buffer
		-- of 500mb. This prevents the device from immediately running back out of space.

		local multiplier = WFPlatformInfo.bytesInKilobyte()

		buffer = 500 * multiplier * multiplier -- 500MB

	end

	return math.max( 0, buffer )

end

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

calculatePurgeBufferStream = function( self, reason )

	return getSizeOfLrCache( self )
	
		:map( function( currentCacheSize )

			local buffer = calculatePurgeBuffer(
					reason, currentCacheSize, self._targetCacheSize )

			if buffer > 0 then
				logInfof( self, "Using a purge buffer of %s.", storageByteString( buffer ) )
			end
	
			return buffer

		end )

end

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

local function getDiskSpaceNeeded( self, maybeExtraSpaceNeeded, maybeHasClearedDeletionBacklog )

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

	-- spaceNeededToPreventNoDiskSpace is the space needed to prevent a no-disk-space
	-- scenario.
	
	-- spaceNeededToReachTargetDiskSpace is the ADDITIONAL space needed to reach
	-- the user-specified cache size.

	maybeExtraSpaceNeeded = maybeExtraSpaceNeeded or 0

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

	local fileSystemFreeSpace, totalSpace = getFreeSpace( self )

	if fileSystemFreeSpace == nil then

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

		return Observable.just( 0, 0, 0 )

	end

	-- If Lr breaches the noDiskSpaceThreshold value, then Lr is in a no-disk-space state.
	-- In this scenario, we try to purge binaries until we're out of a no-disk-space scenario.

	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 getSizeOfLrCache( self )

		:flatMap( function( currentCacheSize )

			logInfof( self, "Max Photo Cache Size: %s, Current Photo Cache Size: %s",
					storageByteString( self._targetCacheSize ), storageByteString( currentCacheSize ) )

			logInfof( self, "Min Free Space: %s, Current Free Space: %s",
					storageByteString( noDiskSpaceThreshold ), storageByteString( fileSystemFreeSpace ) )

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

			local spaceNeededToReachTargetDiskSpace = math.max( 0,
					currentCacheSize - self._targetCacheSize - spaceNeededToPreventNoDiskSpace )
			
			maybeLogDiskStats( self, fileSystemFreeSpace, totalSpace, currentCacheSize )
			
			if not maybeHasClearedDeletionBacklog
					and ( spaceNeededToPreventNoDiskSpace > 0 or spaceNeededToReachTargetDiskSpace > 0 )
					and self._deletionBacklog then
		
				-- If we detect that we may need to reclaim space ( reactive purge ), first process all
				-- the deletions in the deletion backlog and then check again.

				return self._deletionBacklog:processDeletionBacklog()
					:flatMapLatest( function()
						return getDiskSpaceNeeded( self, maybeExtraSpaceNeeded, true )
					end )

			end

			local percentOfTargetSpaceUsed = currentCacheSize / self._targetCacheSize

			return Observable.just(
					spaceNeededToPreventNoDiskSpace,
					spaceNeededToReachTargetDiskSpace,
					percentOfTargetSpaceUsed )

		end )

end

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

-- @mark Logic Determining WHEN to Purge

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

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

local function reactivePurgeIsRequiredStream( 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( kDiskSpaceCheckPeriod, kDiskSpaceCheckPeriod, 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 trigger Reactive Purge.." )
			return getDiskSpaceNeeded( self )

		end )
		
		:filter( function(
					spaceNeededToPreventNoDiskSpace,
					spaceNeededToReachTargetDiskSpace,
					percentOfTargetSpaceUsed )

			-- 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
			
				logInfof( self, "The Volume is out of space." )
				shouldContinue = true
				
			else
			
				logInfof( self, "The Volume is NOT out of space." )
				self._isOutOfSpace = false
			
			end
			
			-- Check if we are purging because we want to reach our target disk space.
			
			if spaceNeededToReachTargetDiskSpace > 0 then
			
				logInfof( self, "The Photo Cache is too large." )
			
				shouldContinue = true
				
			elseif not shouldContinue then
			
				logInfof( self, "The Photo Cache is NOT too large." )
				self._isOverTargetDiskSpaceUsage = false
				
			end
		
			updateVolumeStatusStream( self )

			if not shouldContinue then
				logInfof( self, "Reactive Purge is not necessary at this time." )
			end
			
			return shouldContinue
		
		end )
		
end

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

local function ongoingReactivePurge( self )

	-- Reactive purge triggers when either the volume is out of disk space,
	-- or when the photo cache grows beyond the user-specified size.

	return isVolumeOnlineStream( self )

		:flatMapLatest( reactivePurgeIsRequiredStream( self ) )
	
		:filter( function()
		
			if self._purgeInProgress then
			
				logInfof( self, "Purge already in progress." )	
				
				return false
				
			end
			
			return true
		
		end )
	
		: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 )

			local desc
			
			if spaceNeededToPreventNoDiskSpace > 0 then
				desc = 'Reactive Purge has been triggered. Reason: Out of Free Space'
			elseif spaceNeededToReachTargetDiskSpace > 0 then
				desc = 'Reactive Purge has been triggered. Reason: Photo Cache is too big'
			end

			return performPurge( self, nil, spaceNeededToPreventNoDiskSpace, spaceNeededToReachTargetDiskSpace, desc )
			
				:map( function( 
						spaceNeededAfterPurgeToPreventNoDiskSpace,
						spaceNeededAfterPurgeToReachTargetDiskSpace )
				
					return spaceNeededToPreventNoDiskSpace,
							spaceNeededToReachTargetDiskSpace,
							spaceNeededAfterPurgeToPreventNoDiskSpace,
							spaceNeededAfterPurgeToReachTargetDiskSpace
				
				end )
			
		end )

		:map( function(
					spaceNeededToPreventNoDiskSpace,
					spaceNeededToReachTargetDiskSpace,
					spaceNeededAfterPurgeToPreventNoDiskSpace,
					spaceNeededAfterPurgeToReachTargetDiskSpace )
		
			if spaceNeededAfterPurgeToPreventNoDiskSpace > 0 then

				self._isOutOfSpace = true
				
				logWarnf( self, "Reactive Purge Failed. Still Out of Free Space. Still need %s.", storageByteString( spaceNeededAfterPurgeToPreventNoDiskSpace ) )
				logWarnf( self, "We are now in a NO DISK SPACE state." )
			
			else
			
				self._isOutOfSpace = false
			
			end
			
			if spaceNeededAfterPurgeToReachTargetDiskSpace > 0 then
			
				self._isOverTargetDiskSpaceUsage = true

				logInfof( self, "Reactive Purge Failed. Photo Cache is still too big. Still need %s.", storageByteString( spaceNeededAfterPurgeToReachTargetDiskSpace ) )
			
			else
			
				self._isOverTargetDiskSpaceUsage = false
			
			end
			
			if not self._isOutOfSpace and not self._isOverTargetDiskSpaceUsage then

				local spaceReclaimed = spaceNeededToPreventNoDiskSpace
					+ spaceNeededToReachTargetDiskSpace
					- spaceNeededAfterPurgeToPreventNoDiskSpace
					- spaceNeededAfterPurgeToReachTargetDiskSpace

				logInfof( self, "Reactive Purge Succeeded! Space Reclaimed: %s",
						storageByteString( spaceReclaimed ) )

				logInfof( self, "Reactive Purge Succeeded" )
				
			end
		
			updateVolumeStatusStream( self )
		
			self._debugReactivePurgeRunComplete:onNext( {
				spaceNeededBeforePurgeToPreventNoDiskSpace = spaceNeededToPreventNoDiskSpace,
				spaceNeededBeforePurgeToReachTargetDiskSpace = spaceNeededToReachTargetDiskSpace,
				spaceNeededAfterPurgeToPreventNoDiskSpace = spaceNeededAfterPurgeToPreventNoDiskSpace,
				spaceNeededAfterPurgeToReachTargetDiskSpace = spaceNeededAfterPurgeToReachTargetDiskSpace
			} )
		
			return true
		
		end )

		:catch( function( errorMessage )
		
			self._purgeInProgress = false
			self._purgeInProgressStream:onNext( self._purgeInProgress )
			
			updateVolumeStatusStream( self )
			
			return self._reactivePurgeTask:throwIfNotStopped( errorMessage )
			
		end )
		
		:finally( function()
			updateVolumeStatusStream( self )
		end )

end

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

local function waitUntilIsReadyForPurgeOperation( self )
		
	-- Wait until the volume is online, and no reactive purge is running.

	return Observable.combineLatest(
			isVolumeOnlineStream( self ),
			self._purgeInProgressStream )

		:filter( function( isVolumeOnline, purgeInProgress )
		
			return isVolumeOnline and not purgeInProgress
			
		end )
		
		:take( 1 )

end

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

local function triggerProactivePurge( self, reason )

	local noopResult = {
		reason = reason,
		spaceToAttemptToReclaimProactively = 0,
		spaceReclaimed = 0
	}

	local result

	if reason == purgeReasons.PROACTIVE_PROXIES and self._keepAllProxies then
	
		result = Observable.just( noopResult )
		
	elseif reason == purgeReasons.PROACTIVE_ORIGINALS and self._keepAllOriginals then
	
		result = Observable.just( noopResult )
		
	else

		result = waitUntilIsReadyForPurgeOperation( self )
		
			:flatMapLatest( calculatePurgeBufferStream( self, reason ) )
			
			:flatMapLatest( function( spaceToAttemptToReclaimProactively )

				logInfof( self, "Starting Proactive Purge: %s", purgeReasonStrings[ reason ] )

				if spaceToAttemptToReclaimProactively > 0 then
				
					local desc = string.format( 'proactive_purge_%s', purgeReasonStrings[ reason ] )
				
					return performPurge( self, reason, 0, spaceToAttemptToReclaimProactively, desc )
					
						:flatMap( function( _, spaceNeededAfterPurgeToReachTargetDiskSpace )
					
							local spaceReclaimed = spaceToAttemptToReclaimProactively 
									- spaceNeededAfterPurgeToReachTargetDiskSpace
					
							return logProactivePurgeResults( self, spaceReclaimed, reason )
					
								:map( function()
					
									return {
										reason = reason,
										spaceToAttemptToReclaimProactively = spaceToAttemptToReclaimProactively,
										spaceReclaimed = spaceReclaimed
									}
								
								end )
						
						end )
					
				else
				
					logInfof( self, "Proactive Purge %s is not necessary.", purgeReasonStrings[ reason ] )
				
				end

				return Observable.just( noopResult )

			end )
		
	end
		
	return result
		
		:tap( function( ... )
		
			self._debugProactivePurgeRunComplete:onNext( ... )
		
		end )
		
end 

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

local function ongoingProactivePurge( self )

	local proactivePurgeDelayOnStartup = self._debugProactivePurge and 0.01 or 30
		-- Wait 30 seconds before starting proactive purge.
		
	local proactivePurgeFrequency = self._debugProactivePurge and 10 or daysToSeconds( kProactivePurgeFrequencyInDays )
		-- Every 24 hours, run proactive purge.
		
	local proactivePurgeStream = Observable.timer( proactivePurgeDelayOnStartup, WFRxScheduler.sharedScheduler() )

		:take( 1 )
		
		:flatMap( WFRxPersistentTimer( self._docStore, 'ongoingProactivePurge', proactivePurgeFrequency ) )

		-- Need to throw as proactivePurgeStream doesn't ever complete.
		
		:flatMapLatest( function()
			return self._proactivePurgeTask:throwIfNotRunning()		
		end )

	if WFPlatformInfo.isMobile() then
	
		-- On Mobile, proactive purge will attempt to purge all inactive, non-rendition
		-- binaries. This includes originals, proxies, develop settings, masks, etc.
	
		proactivePurgeStream = proactivePurgeStream
		
			:flatMapLatest( triggerProactivePurge( self, purgeReasons.PROACTIVE_ALL ) )
	
	else
	
		-- On Desktop, proactive purge will attempt to purge all inactive originals.
	
		proactivePurgeStream = proactivePurgeStream
		
			:flatMapLatest( triggerProactivePurge( self, purgeReasons.PROACTIVE_ORIGINALS ) )
			
			:tap( function()
			
				logInfof( self, "Proactive Purge Complete" )
			
			end )
	
	end

	return proactivePurgeStream
	
		:catch( function( errorMessage )
			self._debugProactivePurgeRunComplete:onNext( {} )
			return self._proactivePurgeTask:throwIfNotStopped( errorMessage )
		end )

end

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

-- @mark WFCatVolumeAgent APIs

--[[

	WFCatVolumeAgent; A Guide.

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

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

	For Lightroom Desktop, WFCatVolumeAgent is responsible for purging originals and proxies.
	Purging of renditions and previews is handled by the pixel provider.

	For Mobile, previews and renditions are purged by the purge agent.
	
	It should be noted that WFCatVolumeAgent also handles the purging of binaries
	from disk when an asset is deleted from Lr. This process is unrelated
	to the ongoing purge task. See _purgeOnDeleteSubscription.

	Purge operates differently depending on the type of volume, the
	target disk space threshold, and what's being stored on the volume.
		
	----------------------------------------------------------------------------
		
	Primary Volume
		
		Default (Managed)
		
		Lr manages disk space on the primary volume by purging binaries from
		disk when we detect they are no longer important to the user.
		
		LrD-Only: Keep All Originals
	
		With this option enabled, Lr will never remove originals from disk unless they
		are explicitly deleted by the user.
		
		LrD-Only: Keep All Proxies
		
		With this option enabled, Lr will never remove proxies from disk unless they
		are explicitly deleted by the user.
		
	Secondary Volume ( LrD-Only )
		
		If a secondary volume is being used to store negatives, we only purge negatives
		reactively. Proactive purge is disabled on secondary volumes.
		
		Keep All Originals:
	
		With this option enabled and a secondary volume selected, Lr will never remove
		originals from disk unless they are explicitly deleted by the user.
		
		Keep All Proxies:
		
		With this option enabled and a secondary volume selected, Lr will never remove
		proxies from disk unless they are explicitly deleted by the user.

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

	When to Purge:

	Proactive Purge - Remove some unimportant 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 Lr detects that the cache
	is too big, and when the volume is out of disk space.
		
]]--

local WFCatVolumeAgent = {

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

	_catalog = false,
		-- (WFCatalog)

	_docStore = false,
		-- (WFRxDocStore)

	_indexStore = false,
		-- (WFRxIndexStore)

	_dataCache = false,
		-- (KSDataCache) LrA-Only

	_idleDownloader = false,
		-- (WFCatIdleDownloader)
		
	_deletionBacklog = false,
		-- (NmbDeletionBacklog) LrD-Only
		
	_volumeRootPath = false,
		-- (string) The partial path to where Lr binaries are stored on the volume.

	_volumeGuid = false,
		-- (string) LrD-Only: 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.

	_binaryStrategy = false,
		-- (string) LrM-Only: the binary download strategy being used.

	_targetCacheSize = false,
		-- (number) The target cache size for Lr. AKA the max size of the Lr cache.

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

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

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

	_purgeOnDeleteSubscription = 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.

	_volumeStatusCheckPeriod = false,
		-- (number) How often to check the connection status of the volume.
		
	_isOverTargetDiskSpaceUsage = false,
		-- (boolean) Track whether or not we are over the target disk space usage.

	_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.
		
	_protectedFromPurge = false,
		-- (table) A list of assets to protect from being purged.
		-- key: asset id
		-- value: ref count
		
	_protectedRingBuffer = false,
		-- (WFRingBuffer) A ring buffer of recently accessed assets that are protected from purge.
		
	_volumeConnectionSubscription = false,
		-- (Rx Subscription) A subscription to an observable that periodically checks the volumes connection status.
		
	_untrackedBinariesRemoverSubscription = false,
		-- (Rx Subscription) A subscription to and observable that deletes untracked binaries from
		-- disk and removes the doc value local only annotations.
		
	_hasSentDiskStats = false,
		-- (bolean)
		
	_remainingSpace = false,
		-- (number) The amount of remaining free space on the volume, minus the reserved space.
		
	_isBinaryMarkedForPurgeByTypeAndLocalId = false,

	_presetProfileDeletionStream = false,
		-- (Rx Subject) Fires each time a profile is deleted successfully.
		
	_debugReactivePurgeRunComplete = false,
		-- (Rx Subject) Fires each time a reactive purge run completes. Only used for debugging and tests.
		
	_debugProactivePurgeRunComplete = false,
		-- (Rx Subject) Fires each time a proactive purge run completes. Only used for debugging and tests.

	_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. Used in testing.
		
	_debugMockLrSize = false,
		-- (number) If set, mocks the total size of Lr. Used in testing.
		
	_debugLogRingBuffer = false,
		-- (WFRingBuffer) A ring buffer of logging output that is included in the diagnostic log.

	_debugPurgeListStream = false,
		-- (Subject)  Used in testing.
		
	WFCloseSupport.addCloseSupportMethods,

	private = { -- The following are methods are internal, but are exposed for testing.
		
		-- Misc
		_purgeReasons = purgeReasons,
		_daysToSeconds = daysToSeconds,
		_isVolumeOnlineStream = isVolumeOnlineStream,
		
		-- HOW MUCH to purge
		_getFreeSpace = getFreeSpace,
		_calculatePurgeBuffer = calculatePurgeBuffer,
		_getDiskSpaceNeeded = getDiskSpaceNeeded,
		
		-- Purge Algorithm
		_isProtectedFromPurge = isProtectedFromPurge,
		_getBinaryTypesToPurge = getBinaryTypesToPurge,
		_purgeToReclaimSpace = purgeToReclaimSpace,
		_processBatchOfBinariesForPurge = processBatchOfBinariesForPurge,
		_fetchBinariesToPurge = fetchBinariesToPurge,
		_performPurge = performPurge,
		
		-- WHEN to purge
		_reactivePurgeIsRequiredStream = reactivePurgeIsRequiredStream,
		
	},
	
	class = {
		checkAvailableSpaceOnVolume = checkAvailableSpaceOnVolume,
		purgeReasons = purgeReasons
	}
	
}

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

function WFCatVolumeAgent:__init( catalog, volumeRootPath, volumeGuid, 
		optIdleDownloader, optDataCache, keepLocalInfo, optOverrides )

	check( 'WFCatVolumeAgent', 'WFCatalog', '+string', '?+string', 
			'?WFCatIdleDownloader', '?KSDataCache', '?table', '?table' )

	optOverrides = optOverrides or {}

	self._catalog = catalog
	
	self._docStore = catalog:docStore()

	self._indexStore = catalog:indexStore()

	self._dataCache = WFPlatformInfo.isAndroid and optDataCache or nil

	self._idleDownloader = optIdleDownloader
	
	self._deletionBacklog = WFPlatformInfo.isDesktop and catalog:getDeletionBacklog() or nil

	self._volumeRootPath = volumeRootPath

	self._volumeGuid = volumeGuid
	
	self._logger = WFLogger( string.format( 'WFCatVolumeAgent:%s', self._volumeGuid or '/' ) )
	
	self._debugLogRingBuffer = WFRingBuffer.make( kMaxDebugLogSize )

	self._protectedRingBuffer = WFRingBuffer.make( kMaxSizeOfProtectedRingBuffer, true )

	self._keepAllOriginals = keepLocalInfo and keepLocalInfo.keepAllOriginals

	self._keepAllProxies = keepLocalInfo and keepLocalInfo.keepAllProxies

	if WFPlatformInfo.isMobile() then
		self._binaryStrategy = optOverrides.binaryStrategy or LrmConfig.StrategyPreferProxy
	else
		self._binaryStrategy = nil
	end

	self._volumeStatusStream = Rx.ReplaySubject( 1 )
	
	self._isSyncPausedStream = Rx.BehaviorSubject()
	
	self._hasInternetConnectionStream = Rx.BehaviorSubject()
	
	self._debugReactivePurgeRunComplete = Rx.Subject()
	
	self._debugProactivePurgeRunComplete = Rx.Subject()
	
	self._debugPurgeListStream = Rx.Subject()
	
	self._protectedFromPurge = {}
	
	self._isBinaryMarkedForPurgeByTypeAndLocalId = {}
	
	self._purgeInProgressStream = Rx.BehaviorSubject( false )	

	self._remainingSpace = 0
	
	local purgeOnDeleteStream, presetProfileDeletionStream
			= WFCatPurgeOnDelete.setupPurgeOnDelete( self._catalog, self._deletionBacklog, self._dataCache )
	
	self._purgeOnDeleteSubscription = purgeOnDeleteStream:subscribe()
	
	self._presetProfileDeletionStream = presetProfileDeletionStream
	
	-- Purge settings

	self._volumeStatusCheckPeriod = WFPlatformInfo.isDesktop and 5 or 0
		-- How often to check the connection status of the volume. Mobile
		-- should not have to worry about disconnected volumes, so we set this to
		-- to 0 to indicate no checking is necessary.
		
	if WFPlatformInfo.isDesktop then
	
		self._maxDiskSpacePollingInterval = optOverrides.diskSpacePollingInterval or NmbFileManagementSettings.maxPollingInterval()
		self._minimumFreeSpace = optOverrides.minimumFreeSpace or NmbFileManagementSettings.minimumFreeSpace()
		self._targetCacheSize = optOverrides.targetCacheSize or NmbFileManagementSettings.targetCacheSize()
		
		-- 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
		
	else
	
		self._maxDiskSpacePollingInterval = optOverrides.diskSpacePollingInterval or LrmVolumeAgentSettings.maxPollingInterval()
		self._minimumFreeSpace = optOverrides.minimumFreeSpace or LrmVolumeAgentSettings.minimumFreeSpace()
		self._targetCacheSize = optOverrides.targetCacheSize or LrmVolumeAgentSettings.targetCacheSize()
		
	end
	
	-- If the target cache size is zero, keep a very small cache.
	
	if self._targetCacheSize == 0 then
		self._targetCacheSize = 250 * 1024 * 1024 -- 250mb
	end
	
	-- Purge debug overrides
	
	self._debugProactivePurge = optOverrides.debugProactivePurge or false
	self._debugReactivePurge = optOverrides.debugReactivePurge or false
	self._debugMockDiskFreeSpace = optOverrides.debugMockDiskFreeSpace or false
	self._debugMockLrSize = optOverrides.debugMockLrSize or false

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

	-- Monitor and delete untracked binaries

	self._untrackedBinariesRemoverSubscription 
			= getUntrackedBinariesRemover( self ):subscribe()

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

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

	-- Proactive Purge

	local proactiveTaskDelays = {
		progressiveBackoff = false,
		errorDelay = 5,
		maxDelay = 60,
	}
	
	local proactiveTask = function()
	
		if optOverrides.disableProactivePurge or self._debugReactivePurge or DISABLE_PURGE 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( "WFCatVolumeAgent:Proactive", proactiveTask, proactiveTaskDelays )

	WFDiagnosticLog.addLogProvider( self )

end

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

function WFCatVolumeAgent:close()

	check( 'WFCatVolumeAgent' )
	
	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 purgeOnDeleteSubscription = self._purgeOnDeleteSubscription
	self._purgeOnDeleteSubscription = nil
	
	if purgeOnDeleteSubscription then
		purgeOnDeleteSubscription: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 untrackedBinariesRemoverSubscription = self._untrackedBinariesRemoverSubscription
	self._untrackedBinariesRemoverSubscription = nil
	
	if untrackedBinariesRemoverSubscription then
		untrackedBinariesRemoverSubscription:dispose()
	end

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

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

	local debugPurgeListStream = self._debugPurgeListStream
	self._debugPurgeListStream = nil
	
	if debugPurgeListStream then
		debugPurgeListStream:dispose()
	end

	local presetProfileDeletionStream = self._presetProfileDeletionStream
	self._presetProfileDeletionStream = nil	
	
	if presetProfileDeletionStream then
		presetProfileDeletionStream:dispose()
	end

end

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

--[[!
	\brief Updates the target cache size.
	\return Observable<boolean> Fires with true once complete.
--]]

function WFCatVolumeAgent:setPhotoCacheSize( targetCacheSize )

	check( 'WFCatVolumeAgent', 'number' )
		
	self._targetCacheSize = targetCacheSize
	
	if NmbFileManagementSettings then
		NmbFileManagementSettings.setTargetCacheSize( targetCacheSize )
	elseif LrmVolumeAgentSettings then
		LrmVolumeAgentSettings.setTargetCacheSize( targetCacheSize )
	end
	
	return Observable.just( true )
	
end

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

--[[!
	\brief Get the maximum size of the photo cache.
	\return Observable<number> Size in bytes.
--]]

function WFCatVolumeAgent:getMaximumPhotoCacheSize()

	check( 'WFCatVolumeAgent' )
		
	local size
		
	if NmbFileManagementSettings then
		size = NmbFileManagementSettings.targetCacheSize()
	elseif LrmVolumeAgentSettings then
		size = LrmVolumeAgentSettings.targetCacheSize()
	end
	
	return Observable.just( size )
	
end

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

--[[!
	\brief LrM: Updates the binary strategy used by the application.
	\return Observable<boolean> Fires with true once complete.
--]]

function WFCatVolumeAgent:setBinaryStrategy( binaryStrategy )

	check( 'WFCatVolumeAgent', '+string' )
		
	self._binaryStrategy = binaryStrategy
	
	return Observable.just( true )
	
end

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

--[[!
	\brief Stops the purge agent.
	\return Observable<boolean> Fires with true once complete.
--]]

function WFCatVolumeAgent:start()

	check( 'WFCatVolumeAgent' )

	self._reactivePurgeTask:startContinuousTask()
	self._proactivePurgeTask:startContinuousTask()
	
	return Observable.just( true )
	
end

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

--[[!
	\brief Stops the purge agent.
	\return Observable<boolean> Fires with true once complete.
--]]

function WFCatVolumeAgent:stop()

	check( 'WFCatVolumeAgent' )

	self._reactivePurgeTask:stopContinuousTask()
	self._proactivePurgeTask:stopContinuousTask()
	
	return Observable.just( true )
	
end

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

--[[!

	\brief Ask the volume agent to provide n bytes of free space.
	
	\param desc (string) description of operation to use for debugging and diagnostic logs
	\param bytes (number) The number of bytes needed.
 	\param forcePurge (boolean) If true, this API will purge binaries to make
		space, even if the resulting amount of free space is less than
		the requested bytes. 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 the requested bytes.
	
	\return (number) The amount of free space, in bytes, that can be provided.

--]]

function WFCatVolumeAgent:tryToClaimSpace( desc, bytes, forcePurge )

	check( 'WFCatVolumeAgent', 'string', 'number', '?boolean' )
	
	return isVolumeOnlineStream( self )
	
		:take( 1 )
	
		:flatMapLatest( function( isVolumeOnline )
		
			if not isVolumeOnline then
				return Observable.just( 0 )
			end
		
			return getDiskSpaceNeeded( self, bytes )

				:flatMapLatest( function( spaceNeededToPreventNoDiskSpace )
				
					if spaceNeededToPreventNoDiskSpace == 0 then
					
						return Observable.just( bytes )
							-- If we don't need to reclaim any space, we're done.
							
					end
				
					-- Attempt to reclaim the needed space.
				
					return self._purgeInProgressStream
			
						:filter( function( purgeInProgress )
						
							if not purgeInProgress then
							
								return true
							
							end
							
							logInfof( self, "Purge already in progress. Waiting..." )	
							
							return false
							
						end )
						
						:take( 1 )
						
						:flatMapLatest( function()
						
							local maybePurgeReason = not forcePurge and purgeReasons.NO_DISK_SPACE_ALL_OR_NOTHING
						
							return performPurge( self, maybePurgeReason, spaceNeededToPreventNoDiskSpace, 0, desc )
							
						end )
						
						:map( function( spaceNeededAfterPurgeToPreventNoDiskSpace )
						
							return bytes - math.max( 0, spaceNeededAfterPurgeToPreventNoDiskSpace )
							
						end )
				
				end )
		end )
		
end

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

--[[!

	\brief Force the purge agent to purge all purgeable binaries within the 
			photo cache.
	
	\return Observable<true> Fires with true upon completion.
	
--]]

function WFCatVolumeAgent:clearCache()

	check( 'WFCatVolumeAgent' )

	local bytesToReclaim = 10 * 1024 * 1024 * 1024 * 1024 -- Reclaim all the bytes!
		-- 10 terabytes
		
	return self._purgeInProgressStream

		:filter( function( purgeInProgress )
		
			if not purgeInProgress then
			
				return true
			
			end
			
			logInfof( self, "Purge already in progress. Waiting..." )	
			
			return false
			
		end )
		
		:take( 1 )
		
		:flatMapLatest( function()
		
			local reason = purgeReasons.EMPTY_CACHE
		
			local desc = 'Clear Cache'
		
			return performPurge( self, reason, bytesToReclaim, 0, desc )
			
		end )
		
		:map( function()
		
			return true
			
		end )

end

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

--[[!

	\brief Adds an asset to a ring buffer. Entries within the ring buffer are
		protected from purge. The ring buffer has a fixed size, and the oldest
		entries are cleared when the ring buffer grows too large.
		
		The primary use case for this API is to protect recently accessed and/or
		downloaded assets from purge.
	
	\param assetId (+string|number) The asset doc Id or local Id.

	\return Observable<boolean> Fires with true once complete.

--]]

function WFCatVolumeAgent:addToProtectedRingBuffer( assetId )

	check( 'WFCatVolumeAgent', '+string|number' )

	if not WFRingBuffer.hasItem( self._protectedRingBuffer, assetId ) then
	
		WFRingBuffer.addItem( self._protectedRingBuffer, assetId )
	
	end

	return Observable.just( true )

end

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

--[[!

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

	\return Observable<boolean> Fires with true once complete.

--]]

function WFCatVolumeAgent:protectFromPurge( assetIds )

	check( 'WFCatVolumeAgent', 'table' )

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

	return Observable.just( true )

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 Observable<boolean> Fires with true once complete.

--]]

function WFCatVolumeAgent:releaseProtectionFromPurge( assetIds )

	check( 'WFCatVolumeAgent', '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

	return Observable.just( true )

end

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

--[[

	\brief Checks if an assetId is locked

	\param assetId (string) A asset id or local id.

	\return bool

]]
function WFCatVolumeAgent:isLocked( assetId )

	return self._protectedFromPurge[ assetId ] and true or false

end

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

--[[!

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

--]]

function WFCatVolumeAgent:getVolumeStatusStream()

	check( 'WFCatVolumeAgent' )
	return self._volumeStatusStream

end

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

--[[!

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

--]]

function WFCatVolumeAgent:getVolumeGuid()

	check( 'WFCatVolumeAgent' )
	return self._volumeGuid

end

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

--[[!

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

--]]

function WFCatVolumeAgent:isSyncPausedStream()

	check( 'WFCatVolumeAgent' )
	return self._isSyncPausedStream

end

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

--[[!

	\brief Sets the sync-paused status.
	\param setSyncPausedStatus (boolean)
	\return void

--]]

function WFCatVolumeAgent:setSyncPausedStatus( isSyncPaused )

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

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

--[[!

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

--]]

function WFCatVolumeAgent:hasInternetConnectionStream()

	check( 'WFCatVolumeAgent' )
	return self._hasInternetConnectionStream

end

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

--[[!

	\brief Sets the internet connection status.
	\param hasInternetConnection (boolean)
	\return void

--]]

function WFCatVolumeAgent:setInternetConnectionStatus( hasInternetConnection )

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

end

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

--[[!

	\brief Returns a stream that fires when a profile is deleted successfully.
	\return (Rx Subject)

--]]

function WFCatVolumeAgent:getPresetProfileDeletionStream()

	check( 'WFCatVolumeAgent' )
    return self._presetProfileDeletionStream

end

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

--[[!

	\brief Purge an external develop settings file from disk if it has no
		references. Note that this function is only implemented for LrA; LriOS
		does not currently share develop settings between files (and therefore
		does not need to reference count them). On LrD, purging of unreferenced
		develop settings is handled by the pixel provider.
	
	\param externalDevelopSettingsFilename (string) File name.

	\return Observable<boolean> Fires with true.

--]]

function WFCatVolumeAgent:maybePurgeExternalDevelopSettingsFile( externalDevelopSettingsFilename )

	check( 'WFCatVolumeAgent', '+string' )

	if WFPlatformInfo.isMobile() then

		return LrmOzPurgeAgentUtils.maybePurgeXmpDevelopFiles(
				self._catalog, { externalDevelopSettingsFilename } )

	end

	return Observable.just( true )

end

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

--[[!

	\brief Generates diagnostic log data for the volume agent.
	\return void

--]]

function WFCatVolumeAgent:sendDiagnosticLogData()

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

	local numberOfProxiesInProxiesFolder, numberOfOriginalsInOriginalsFolder
	local sizeOfProxiesFolder, sizeOfOriginalsFolder

	local catalogPath = self._catalog:fileSpec():catalogPath()

	local catalogPathParent = KSPathUtils.parent( catalogPath )

	local proxiesFolder = child( catalogPath, 'proxies' )

	local originalsFolder

	if WFPlatformInfo.isDesktop then
	
		local originalsPathInfo = self._catalog:fileSpec():originalsPathInfo() or {}

		local success, originalsRootPath = pcall( 
				WFFileUtils.createFullPathFromRelativePathAndGuid,
						originalsPathInfo.relativePath, originalsPathInfo.volumeGuid )

		if success and type( originalsRootPath ) == 'string' then

			originalsFolder = child( originalsRootPath, 'originals' )

		else
		
			originalsFolder = child( catalogPath, 'originals' )
			
		end
	
	else
	
		originalsFolder = child( catalogPath, 'originals' )
		
	end
	
	local sizeOfCatalogFolder = WFAsyncFileUtils.calculatePathSize( catalogPath )
		:map( function( bytes )
			return storageByteString( bytes )
		end )

	local sizeOfLrFolder = WFAsyncFileUtils.calculatePathSize( catalogPathParent )
		:map( function( bytes )
			return storageByteString( bytes )
		end )

	if WFFileUtils.exists( proxiesFolder ) then
	
		sizeOfProxiesFolder = WFAsyncFileUtils.calculatePathSize( proxiesFolder )
			:map( function( bytes )
				return storageByteString( bytes )
			end )
			
	end

	if WFFileUtils.exists( originalsFolder ) then
	
		sizeOfOriginalsFolder = WFAsyncFileUtils.calculatePathSize( originalsFolder )
			:map( function( bytes )
				return storageByteString( bytes )
			end )
			
	end

	if not WFPlatformInfo.isAndroid then

		-- WFFolderWalker is not available on LrA

		local WFFolderWalker = require 'WFFolderWalker'

		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

		if WFFileUtils.exists( proxiesFolder ) then

			numberOfProxiesInProxiesFolder
					= countFilesInDirectory( proxiesFolder )

		end

		if WFFileUtils.exists( originalsFolder ) then

			numberOfOriginalsInOriginalsFolder
					= countFilesInDirectory( originalsFolder )

		end

	end

	local 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' ] = storageByteString( self._remainingSpace ),
		[ 'HD: Number of Local Originals in Originals Folder' ] = numberOfOriginalsInOriginalsFolder,
		[ 'HD: Number of Local Proxies in Proxies Folder' ] = numberOfProxiesInProxiesFolder,
		[ 'HD: Size of Originals Folder' ] = sizeOfOriginalsFolder,
		[ 'HD: Size of Proxies Folder' ] = sizeOfProxiesFolder,
		[ 'HD: Size of Catalog Folder' ] = sizeOfCatalogFolder,
		[ 'HD: Size of Lr Folder (Catalog Parent)' ] = sizeOfLrFolder,
		
		-- The following metrics are gathered from the Volume Agent:

		[ 'Calculated Storage Buckets' ] = WFCatLocalStorage.getLocalStorageSize( self._catalog )
				
				:map( function( permutations )
				
					local buckets
				
					if WFPlatformInfo.isDesktop then
					
						local storeOriginalsOnMainVolume = WFFileUtils.getVolumeGuidForPath( originalsFolder )
								== WFFileUtils.getVolumeGuidForPath( catalogPath )
					
						buckets = KSTableUtils.derefOpt( 
								permutations, storeOriginalsOnMainVolume, self._keepAllOriginals, self._keepAllProxies )
				
					else
					
						buckets = permutations
					
					end
				
					local formattedBuckets = {}
					
					for key, value in pairs( buckets ) do
					
						formattedBuckets[ key ] = storageByteString( value )
					
					end
					
					return formattedBuckets
				
				end ),

		[ 'Volume GUID' ] = self._volumeGuid,

		[ 'Is storing all originals locally?' ] = self._keepAllOriginals,
		[ 'Is storing all proxies locally?' ] = self._keepAllProxies,
		[ 'Maximum Photo Cache Size' ] = storageByteString( self._targetCacheSize ),

		[ 'Is Photo Cache too large?' ] = self._isOverTargetDiskSpaceUsage,
		[ 'Is volume out of space?' ] = self._isOutOfSpace,

		[ 'WF Purge Log' ] = WFRingBuffer.map( justReturn, self._debugLogRingBuffer ),

	}

	local binaryTypes = getBinaryTypesToPurge()

	for _, binaryType in ipairs( binaryTypes ) do

		body[ string.format( "Total number of local %s", binaryType ) ]
				= WFCatVolumeAgentQueryHelper._DEBUG_numberOfLocalBinariesOfType(
						self._catalog, binaryType )
		
		body[ string.format( "Total size of local %s", binaryType ) ]
				= WFCatVolumeAgentQueryHelper._DEBUG_sizeOfLocalBinariesOfType(
						self._catalog, binaryType )
						
					:map( function( bytes )
						return storageByteString( bytes )
					end )

	end

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

	local uniqueId = KSIdGenerator.uuidInLowercaseHex()

	return {

		name = string.format( "Volume Agent:%s", self._volumeGuid or uniqueId ),
		id = string.format( 'WFCatVolumeAgent:%s', self._volumeGuid or uniqueId ),
		weight = 510,
		parentSection = WFDiagnosticLog.Sections.LOCAL_STORAGE,
		body = body

	}

end

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

-- @mark DEBUG APIs

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

--[[!

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

--]]

function WFCatVolumeAgent:_DEBUG_reactivePurgeRunCompleteStream()

	check( 'WFCatVolumeAgent' )
	return self._debugReactivePurgeRunComplete

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 WFCatVolumeAgent:_DEBUG_proactivePurgeRunCompleteStream()

	check( 'WFCatVolumeAgent' )
	return self._debugProactivePurgeRunComplete

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 WFCatVolumeAgent:_DEBUG_purgeListStream()

	check( 'WFCatVolumeAgent' )
	return self._debugPurgeListStream

end

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

return Rx.Class 'WFCatVolumeAgent'( WFCatVolumeAgent )
