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

LrmOzPurgeAgent.lua


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

ADOBE CONFIDENTIAL
------------------
Copyright 2019 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.

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

This class monitors the content of a single catalog's data cache and periodically
purges it.

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

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

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

	-- WF.core.os
local WFFileUtils = require 'WFFileUtils'
local KSPathUtils = require 'KSPathUtils'

	-- WF.core.rxlua
local WFArguments = require 'WFArguments'

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'
local WFCloseSupport = require 'WFCloseSupport'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFCapabilities = require 'WFCapabilities'
local WFOzSyncConstants = require 'WFOzSyncConstants'

local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'
local WFRxThrottledBatchedBulkDocValue = require 'WFRxThrottledBatchedBulkDocValue'

	-- WF.os.prefs
local WFPlatformInfo = require "WFPlatformInfo"

	-- WF.oz.purge_agent
local WFOzPurgeSettings = require 'WFOzPurgeSettings'

	-- WF.core.rxlua
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- WF.lrmobile.modules.source
local LrmConfig = require 'LrmConfig'
local WFCatSyncAutoDeleter = require 'WFCatSyncAutoDeleter'
local LrmOzPurgeAgentUtils = require 'LrmOzPurgeAgentUtils'

	-- WF.oz_client.utils
local OzFilePaths = require 'OzFilePaths'

	-- WF.oz_client.managed_catalog
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatCopies = require 'WFCatCopies'

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

LrmConfig.registerBinaryStrategyCheck()

local check = assert( WFArguments.check )

local WFCatAssetUtils_isValidOriginalOnOz
		= assert( WFCatAssetUtils.isValidOriginalOnOz, "WFCatAssetUtils.isValidOriginalOnOz was not found" )

local runUntilDone = WFRxRunUntilDone

local defaultCachePollingInterval = 40	-- seconds; feel free to tune

local minimumFreeSpace = WFOzPurgeSettings.minimumFreeSpace()

local logger = WFLogger 'LrmOzPurgeAgent'
	 --:setLevel 'DEBUG'

local derefOpt = KSTableUtils.derefOpt

local batchSize = 100

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

local cacheReferencesPathKey = LrmOzPurgeAgentUtils.cacheReferencesPathKey
local removeCacheEntriesAndXmpsForDocument = assert( LrmOzPurgeAgentUtils.removeCacheEntriesAndXmpsForDocument )
local absolutePath = assert( LrmOzPurgeAgentUtils.absolutePath )
local purgeBinariesDirectly = assert( LrmOzPurgeAgentUtils.purgeBinariesDirectly )
local userUpdatedKey = assert( LrmOzPurgeAgentUtils.userUpdatedKey )
local maybePurgeXmpDevelopFiles = assert( LrmOzPurgeAgentUtils.maybePurgeXmpDevelopFiles )

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

local getDiskSpaceNeeded = function( self )

	local path

	if WFPlatformInfo.isAndroid then
		path = KSPathUtils.parent( self._dataCachePath )
	elseif WFPlatformInfo.isiOS then
		path = OzFilePaths.getCatalogBasePath( self._catalogId )
	end

	local fileSystemFreeSpace, _ -- fileSystemSize
		= WFFileUtils.freeSpaceAtPath( path )

	local requiredSpace = math.max(0, minimumFreeSpace - fileSystemFreeSpace)
	return requiredSpace

end

--============================================================================--
--[[!
	\brief A \c LrmOzPurgeAgent monitors the content of a single catalog's data
	cache and periodically purges it to make additional disk space available.
	
	It should be used only by LrmSyncClientImp.	
--]]

-- @mark LrmOzPurgeAgent

local purgeAgentMethods = {
	
	_cachePollingInterval = defaultCachePollingInterval,
		-- used to calm queries - configurable for tests

	_docStore = false,
		-- (WFRxDocStore)
	
	_wfCatalog = false,
		-- (WFCatalog)
	
	_indexStore = false,
		-- (WFRxIndexStore)

	_dataCache = false,
		-- (KSDataCache) data cache for this doc store (Android only)
	
	_dataCachePath = false,
		-- (string) path to cache (Android only)
	
	_purgeUserUpdated = false,
		-- An IndexStore query listing purgeable assets by userUpdated/epoch
	
	_purgeOnDeleteChain = false,
		-- the reactive chain that is executed on purgeOnDelete (exposed for testing)
	
	_purgeOnDelete = false,
		-- Subscription to the docStore's document deleter
	
	_checkCacheKeysFrom = false,
		-- (false or string) where do we start cross-checking cache keys?

	_warnedMissingEntries = false,
		-- (table) entries that we've warned about (so we don't keep bleating about them)

	_binaryStrategy = false,
		-- (string) LrmConfig
	
	_purgeSub = false,
		-- A subscription to _purgeUserUpdated used to continously purge unused renditions.

	_catalogId = false,
		-- (string) CatalogId

	_deleteFeedCallback = false,
		-- A subscription to send callback when a profile was deleted successfully (Android only)

	_canPurge = false,
	-- A subscription to send callback when a profile was deleted successfully (Android only)

	_initializedSubject = false,
		-- An Observable that sends completed once initializaton is done

	_autoDeleter = false,
		-- (WFCatSyncAutoDeleter)
		
	_lockedAssets = false,
		-- (table) A set of assetIds which are locked wrt purge( Should be excluded from purge )

	private = {

		-- The following slots are internal, but are exposed for testing.
		
		_getDiskSpaceNeeded = getDiskSpaceNeeded,
			-- a function that returns the needed diskspace

		_purgeContinuously = false
			-- An observable that emits the progress of the continuous purge process
	},

	WFCloseSupport.addCloseSupportMethods,

}

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

function purgeAgentMethods:sendDiagnosticLogData()

	return {

		name = "Lrm Purge Agent",
		id = 'LrmOzPurgeAgent',
		weight = 521,

		body = {
			binaryStrategy = self._binaryStrategy,	
		}
	}
	
end

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

local function purgeableAlbumAssetListSql()

	return string.format( [[
		SELECT docId, assetId, renditionName, %s, size 
		FROM cacheReferences 
		WHERE cacheReferences.hasUploaded == 1
			AND cacheReferences.renditionName != 'thumbnail2x'
			AND cacheReferences.assetId IN ( 
				SELECT 
					album_asset.assetId 
				FROM 
					album_asset
				INNER JOIN 
					albums 
				ON 
					album_asset.albumId == albums.docId 
				WHERE 
					album_asset.albumId == ?)
			AND (cacheReferences.assetId NOT IN ( 
					SELECT 
					   album_asset.assetId 
					FROM 
					   album_asset 
					INNER JOIN 
					   albums 
					ON 
						   album_asset.albumId == albums.docId 
					WHERE 
						   albums.useOffline == 1 
					)
				)
		   ORDER BY docId ASC LIMIT %s]],
		   cacheReferencesPathKey, batchSize)
		   
end
 
 --------------------------------------------------------------------------------

 local function purgeUserUpdatedSql()
 
	 return string.format( [[ 
		SELECT docId, assetId, renditionName, %s, size 
		FROM cacheReferences
			WHERE %s < ( SELECT max( %s ) FROM cacheReferences )
				AND cacheReferences.hasUploaded == 1
				AND cacheReferences.renditionName != 'thumbnail2x'
				AND (cacheReferences.assetId NOT IN (
					SELECT 
						album_asset.assetId 
					FROM 
						album_asset 
					INNER JOIN 
						albums 
					ON 
						album_asset.albumId == albums.docId 
					WHERE 
						albums.useOffline == 1 
				)
			)
			ORDER BY %s ASC, size DESC LIMIT %d ]],
			cacheReferencesPathKey, userUpdatedKey, userUpdatedKey, userUpdatedKey, batchSize )
end

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

local function purgeAllSql()

	return string.format( [[ 
		SELECT docId, assetId, renditionName, %s, size 
		FROM cacheReferences 
			WHERE cacheReferences.hasUploaded == 1
				AND cacheReferences.renditionName != 'thumbnail2x'
				AND (cacheReferences.assetId NOT IN (
					SELECT 
						album_asset.assetId 
					FROM 
						album_asset 
					INNER JOIN 
						albums 
					ON 
						album_asset.albumId == albums.docId 
					WHERE 
						albums.useOffline == 1 
				)
			)
			ORDER BY docId ASC LIMIT %d ]],
			cacheReferencesPathKey, batchSize )

end

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

local function purgeIsEnabled( self )

	local isPurgeEnabledForResidualOrOfflineUsers = WFCapabilities.getCapability( "LrmOzPurgeAgent.isPurgeEnabledForResidualOrOfflineUsers" )
	
	isPurgeEnabledForResidualOrOfflineUsers = isPurgeEnabledForResidualOrOfflineUsers == 'true' or false
	
	if isPurgeEnabledForResidualOrOfflineUsers then
		return Rx.Observable.just( true )
	end
	
	local user = self._wfCatalog:ozUser()
	local isValidUser = user and user:userId() ~= WFOzSyncConstants.kGlobalDocumentId
	
	return Rx.Observable.combineLatest( self._canPurge, user:isSubscriptionExpired() )

		:map( function( canPurge, isSubscriptionExpired )

			if isValidUser then

				-- Don't Purge assets in offline or residual mode
				if isSubscriptionExpired or not canPurge then
					return false
				end

			end

			return true

		end)
		
		:distinctUntilChanged()
		
end

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

local function setBinaryStrategy( self, binaryStrategy )

	if self._binaryStrategy ~= binaryStrategy then

		self._binaryStrategy = binaryStrategy

		self._purgeUserUpdated = self._indexStore:query {
			sql = purgeUserUpdatedSql(),
			quiesceUntilUpToDate = true,
			calm = self._cachePollingInterval,
		}

	end

end

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

local function setupPurgeOnDelete( self )

	local purgeAgentChangeCounterLocalDocId = 'LrmOzPurgeAgent-changeCounter'
	
	self._purgeOnDeleteChain = self._docStore:deletedDocsSinceNamedChangeCounter( purgeAgentChangeCounterLocalDocId )

		:flatMap( function( docValues, continueFrom )

			return Rx.Observable.from( docValues )
			
				:flatMap( function( docValue )

					local result = Rx.Observable.just( nil )

					if docValue.type == 'asset' then

						local hasTrashOrRestoreAsset = derefOpt( docValue, '_newId' ) 
						local deletedFromOzWithNewId = derefOpt( docValue, '_serverProvidedMetadata', 'new_id' )

						if hasTrashOrRestoreAsset then
						
							result = Rx.Observable.just( { docValue._id, nil } )
								-- dont purge as it goes to trash
						
						elseif deletedFromOzWithNewId then
							
							local localOnly = {}
							
							if WFPlatformInfo.isiOS then
							
								localOnly.files = {
									master = derefOpt( docValue, '_localOnly', 'files', 'master' ),
									proxy = derefOpt( docValue, '_localOnly', 'files', 'proxy' ),
								}
						
							else
								
								localOnly = {
									originalFilePath = derefOpt( docValue, '_localOnly', 'originalFilePath' ),
									latestOriginal = derefOpt( docValue, '_localOnly', 'latestOriginal' ),
									original = derefOpt( docValue, '_localOnly', 'original' ),
									proxy = derefOpt( docValue, '_localOnly', 'proxy' ),
									files = derefOpt( docValue, '_localOnly', 'files' ),
								}
								
							end
							
							local deletedFromOzDocOverlay = {
								_id = deletedFromOzWithNewId,
								_localOnly = localOnly
							}
							
							-- keep xmp_develop file

							local copyOptions = {
								needToCopyOriginal = false,
								needToCopyProxy = false,
								needToCopyRenditions = false,
								needToCopyExternalDevelopSettings = true,
								needToCopyXmpSidecar = false
							}
							
							local modifiedDeletedFromOzDocOverlay, filesToCopy = WFCatCopies.calculateCopyPathsAndUpdateDocValue( self._wfCatalog, docValue, deletedFromOzDocOverlay, copyOptions )
										
							local maybeCopyExternalXmpFile = Rx.Observable.just( true )
							
							if #filesToCopy > 0 then
							
								-- note, this block is iOS only
								
								local xmpFileInfo = filesToCopy[ 1 ]
								
								local externalXmpPath = xmpFileInfo.srcFilePath
								local externalXmpCopyPath = xmpFileInfo.dstFilePath
								
								if externalXmpPath and externalXmpCopyPath then
									maybeCopyExternalXmpFile = WFCatCopies.copyExternalXmpFile( externalXmpPath, externalXmpCopyPath )
								end
								
							end
							
							
							result = maybeCopyExternalXmpFile
								:map( function() 
									if WFPlatformInfo.isiOS then
										self:purgeDeletedBlobFolder( docValue._id )
									end
								
									return { docValue._id, modifiedDeletedFromOzDocOverlay }
								end )
													
						else
							
							-- if we are here, this is a final delete of the asset
							
							self:purgeDeletedDocument( docValue )

							result = removeCacheEntriesAndXmpsForDocument( self._wfCatalog, self._dataCache, docValue )
								:map( function()
									return { docValue._id, nil }
								end )
							
						end
						
					elseif docValue.type == 'version' then
					
						-- If a version is deleted, this is the final state and all
						-- associated binaries should be deleted.
						
						self:purgeDeletedDocument( docValue )

						result = removeCacheEntriesAndXmpsForDocument( self._wfCatalog, self._dataCache, docValue )						
							:map( function()
								return { docValue._id, nil }
							end )
					
					end
					
					return result
			
				end )
			
				:toArray()

				:flatMap( function( docIdsAndMaybeOverlays )

					local docIds =  {}
					local overlays = {}
					
					-- docIdsAndMaybeOverlays is an array of { docId, maybeOverlay }
					for i, docIdAndMaybeOverlay in ipairs( docIdsAndMaybeOverlays ) do
						docIds[ i ] = docIdAndMaybeOverlay[ 1 ]
						local overlay = docIdAndMaybeOverlay[ 2 ]
						if overlay then
							overlays[ #overlays + 1 ] = overlay
						end
					end

					if next( overlays ) then
					
						return WFRxThrottledBatchedBulkApplyOverlay( 'LrmOzPurgeAgent_purgeOnDelete', self._docStore, overlays, batchSize, 10 )
							:map( function()
								return docIds
							end )
					
					end
					
					return Rx.Observable.just( docIds )

				end )				
				:flatMap( function( docIds )
			
					return self._docStore:updateDeletedDocsSinceNamedChangeCounter(
							purgeAgentChangeCounterLocalDocId, continueFrom )
							
							:map( function()
							
								return docIds
							
							end )
				
				end )

		end )
		
		:filter( function( docIds )
			return #docIds > 0
		end )
		
		:shareReplay( 1 )

	self._purgeOnDelete = self._purgeOnDeleteChain:subscribe()
	
end 

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

--[[!
	\brief Initialize a purge agent for a specific catalog.
	
	\param catalogService (OzCatalogServiceViaWFCatalog) catalog service
	\param binaryStrategy (LrmConfig) prefer proxy or master
--]]

function purgeAgentMethods:__init( catalogService, binaryStrategy, cachePollingInterval, optArgs )

	check( 'LrmOzPurgeAgent', 'OzCatalogService', '?binaryStrategy', '?number', '?table' )
	
	if cachePollingInterval then
		self._cachePollingInterval = cachePollingInterval
	end
	
	optArgs = optArgs or {}
	
	self._canPurge = BehaviorSubject( false )

	self._warnedMissingEntries = {}

	self._initializedSubject = ReplaySubject( 1 )
	
	self._lockedAssets = {}
			
	WFDiagnosticLog.addLogProvider( self )

	if WFPlatformInfo.isAndroid then

		self._deleteFeedCallback = ReplaySubject( 1 )

		Rx.Observable.combineLatest( function( wfCatalog, dataCache )
	
				self._wfCatalog = wfCatalog

				self._docStore = wfCatalog:docStore()
				self._indexStore = wfCatalog:indexStore()

				self._catalogId = wfCatalog:ozCatalogSpec():catalogId()
	
				-- Setup the Auto Deleter for Cascade Deletions
				self._autoDeleter = WFCatSyncAutoDeleter( wfCatalog, optArgs, dataCache )
				self:addCloseTarget( self._autoDeleter )
							
				self._dataCache = dataCache
				self:addCloseTarget( self._dataCache )
		
				local documentsDir = KSPathUtils.getStandardFilePath 'carouselDocuments'
	
				local catalogDir = KSPathUtils.child( documentsDir, self._catalogId )
				catalogDir = KSPathUtils.child( catalogDir, "renditions" )
	
				WFFileUtils.mkdir( catalogDir )
	
				self._dataCachePath = KSPathUtils.child( catalogDir, "previewCache" )

				if binaryStrategy then
					setBinaryStrategy( self, binaryStrategy )
				end
				
				setupPurgeOnDelete( self )

				self._initializedSubject:onNext( self._catalogId )
				self._initializedSubject:onCompleted()

				return wfCatalog
				
			end,
			catalogService:getWfCatalog(),
			catalogService:getDataCache() )
			:call( runUntilDone )

	elseif WFPlatformInfo.isiOS then
		
		catalogService:getWfCatalog()
			:map( function( wfCatalog )

				self._wfCatalog = wfCatalog

				self._docStore = wfCatalog:docStore()
				self._indexStore = wfCatalog:indexStore()

				self._catalogId = wfCatalog:ozCatalogSpec():catalogId()

				-- Setup the Auto Deleter for Cascade Deletions
				self._autoDeleter = WFCatSyncAutoDeleter( wfCatalog, optArgs )
				self:addCloseTarget( self._autoDeleter )
				
				if binaryStrategy then
					setBinaryStrategy( self, binaryStrategy )
				end

				setupPurgeOnDelete( self )

				self._initializedSubject:onNext( self._catalogId )
				self._initializedSubject:onCompleted()

				return wfCatalog
				
			end )
			:call( runUntilDone )
		
	end
	
end

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

function purgeAgentMethods:getDeleteFeedCallback( )

    return self._deleteFeedCallback

end

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

function purgeAgentMethods:mayBeStartPurging( )

	self._canPurge:onNext( true )

end

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

function purgeAgentMethods:mayBeStopPurging( )

	self._canPurge:onNext( false )

end

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

function purgeAgentMethods:close()

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

	self:stopContinuousPurge()	

end

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

function purgeAgentMethods:purgeDeletedDocument( docValue )

	local hasTrashOrRestoreAsset = derefOpt( docValue, '_newId' ) or derefOpt (docValue, '_serverProvidedMetadata', 'new_id')

	if hasTrashOrRestoreAsset then
		return docValue
	end

	for renditionName, _ in pairs( purgeBinariesDirectly ) do

		local renditionPath = LrmOzPurgeAgentUtils.renditionPath( docValue, renditionName )

		if renditionPath and renditionPath ~= '' then

			WFFileUtils.deleteFile( absolutePath( self._catalogId, renditionPath ) )

			if renditionName == 'presetProfile' and self._deleteFeedCallback then

				self._deleteFeedCallback:onNext( true )

			end

		end

	end

	return docValue

end

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

if WFPlatformInfo.isiOS then

function purgeAgentMethods:purgeDeletedBlobFolder( docId )

	local relativeBlobPath = OzFilePaths.getRelativeBlobPath( docId )
	
	pcall( WFFileUtils.delete, absolutePath( self._catalogId, relativeBlobPath ) )
	
	return docId

end

end

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

local function preparePurgeImp( victims, spaceNeeded, lockedAssets, docValuesMap )

	local result = {
		docs = {},
		debugDocIds = {}, -- Only used by unit tests.
		removeFiles = {},
		removeCache = {},
		developSettingsRelPathsToDelete = {},
		spaceReclaimed = 0,
		purged = 0
	}

	local isUnitTest = WFPlatformInfo.isUnitTest()

	for _, victim in ipairs( victims ) do

		local docId = victim[ 1 ]

		local assetId = victim[ 2 ]
		
		local renditionName = victim[ 3 ]
		
		local docValue = docValuesMap[ docId ]
		
		if not lockedAssets[ assetId ]
				and ( renditionName ~= 'original'
						or ( renditionName == 'original'
								and WFCatAssetUtils_isValidOriginalOnOz( docValue ) ) ) then
			
			logger:debugf('purging docId:%s, assetId:%s', docId, assetId )

			-- Hard-coded indices are based on indexStore queries purgeAllSql, purgeUserUpdatedSql 
			-- and purgeableAlbumAssetListSql

			local overlay = result.docs[ docId ] or {
				_id = docId
			}

			overlay = LrmOzPurgeAgentUtils.updateOverlayWithRenditionName( overlay, renditionName )

			if purgeBinariesDirectly[ renditionName ] then
				result.removeFiles[ #result.removeFiles + 1 ] = victim[ 4 ]
			else
				if renditionName == 'xmp_develop' then
					-- special case for refcounted develop settings on Android
					result.developSettingsRelPathsToDelete[ #result.developSettingsRelPathsToDelete + 1 ] = victim[ 4 ]
				else
					result.removeCache[ #result.removeCache + 1 ] = victim[ 4 ]
				end
			end

			result.docs[ docId ] = overlay
			
			if isUnitTest then
				result.debugDocIds[ #result.debugDocIds + 1 ] = docId
			end
			
			result.spaceReclaimed = result.spaceReclaimed + victim[ 5 ]
			result.purged = result.purged + 1

			if ( spaceNeeded ~= -1 ) and ( result.spaceReclaimed >= spaceNeeded ) then
				return result
			end
			
		else
			logger:debugf('Skipping purge for docId:%s, assetId:%s', docId, assetId )
		end

	end

	return result

end

-- Exposed for testing:

purgeAgentMethods.private._preparePurgeImp = preparePurgeImp

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

--[[

\brief prepare purging of documents

\param victims as an array of { docId, renditionName, relativePath, size }
	   (the output of an index store query)
\param spaceNeeded: spaceNeeded to satisfy minimumFreeSpace threshold or
		-1 for no purge limit

\returns a table with
	- docs:  changed documents
	- removeFiles: files that can be removed
	- removeCache: cache entries to purge
	- spaceReclaimed: space reclaimed
	- developSettingsRelPathsToDelete: an array of xmp develop relative paths (Android only)
	- purged: number of purged documents
]]--

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

local function preparePurge( self, victims, spaceNeeded )

	logger:infof( "preparePurge\nvictims = %s\nspaceNeeded = %s",
				   victims, spaceNeeded )

	local docIds = KSTableUtils.extractColumnAsTable( victims, 1 )

	local assetDocValueStream = 
			WFRxThrottledBatchedBulkDocValue( "preparePurge", self._docStore, docIds, 100, 10 )
			
				:scan( function( docValuesMap, batchOfDocValues )

					for _, docValue in ipairs( batchOfDocValues ) do
						docValuesMap[ docValue._id ] = docValue
					end

					return docValuesMap
					
				end, {} )

				:takeLast( 1 )

	return assetDocValueStream
	
		:map( function( docValuesMap )
		
			return preparePurgeImp( victims, spaceNeeded, self._lockedAssets, docValuesMap )
		
		end )
		
end

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

local function purgeVictims(self, victims, spaceNeeded )

	local purged = 0

	logger:debugf('victims:%s, spaceNeeded:%s', victims, spaceNeeded)

	if not next( victims ) or spaceNeeded == 0 then
		return Rx.Observable.just( 0, spaceNeeded, {} )
	end

	return preparePurge( self, victims, spaceNeeded )

		:flatMap( function( batch )

			if not next ( batch.docs ) then
				return Rx.Observable.just( 0, spaceNeeded, {} )
			end
			
			local overlays = KSTableUtils.extractValues( batch.docs )

			return WFRxThrottledBatchedBulkApplyOverlay( 'LrmOzPurgeAgent_purgeVictims', self._docStore, overlays, batchSize, 10 )
				:takeLast( 1 )
				:flatMap( function()
				
					purged = purged + batch.purged
					
					if spaceNeeded ~= -1 then
						spaceNeeded = math.max(0, spaceNeeded - batch.spaceReclaimed)
					end
				
					-- remove the files
					for i, fileName in ipairs(batch.removeFiles) do
						if fileName and fileName ~= '' then
							WFFileUtils.deleteFile( absolutePath( self._catalogId, fileName ) )
						end
					end
					
					if WFPlatformInfo.isiOS then
						-- no cache entries to purge, everything was in the filesystem
						
						return Rx.Observable.just( purged, spaceNeeded, batch.debugDocIds )
					end

					return self._dataCache:removeBulk( batch.removeCache )
						:flatMap( function()
							return maybePurgeXmpDevelopFiles( self._wfCatalog, batch.developSettingsRelPathsToDelete )
						end )
						:map( function()
							return purged, spaceNeeded, batch.debugDocIds
						end )
						-- purgeVictims is called from a flatMapLatest and can be abandoned, so make sure 
						-- the secondary cleanup is run to completion
						:call( WFRxRunUntilDone )
				end )
			
		end )

end

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

function purgeAgentMethods:purgeUntilDone( query, spaceNeeded )

	return purgeIsEnabled( self )
		:flatMapLatest( function( enabled )

			if not enabled then
				return Rx.Observable.just( 0 )
			else
				return query
					:takeWhile( function( rows )
						return next( rows )
					end )
					:flatMap( function( victims )

						return purgeVictims( self, victims, spaceNeeded )

					end )
					:map( function ( purged, _spaceNeeded )

						spaceNeeded = _spaceNeeded

						return purged

					end )
					:sum()
					:defaultIfEmpty( 0 )
			end
		end )
		:take( 1 )

end

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

--[[!
	\brief Triggers an immediate purge cycle.
]]--

function purgeAgentMethods:purgeImmediately( binaryStrategy, noPurgeLimit )

	check( 'LrmOzPurgeAgent', '?binaryStrategy', '?boolean' )

	if binaryStrategy then
		setBinaryStrategy( self, binaryStrategy )
	end

	local purgeAll = self._indexStore:query{
		sql = purgeAllSql(),
		quiesceUntilUpToDate = true
	}

	return self._initializedSubject
		:flatMap( function()

			-- wait until initialized before calling _getDiskSpaceNeeded

			local spaceRequired = self.private._getDiskSpaceNeeded( self )
			if noPurgeLimit then
				spaceRequired = -1
			end

			return self:purgeUntilDone( purgeAll, spaceRequired )
		end )

end

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

function purgeAgentMethods:purgeForAlbum( albumId, binaryStrategy, noPurgeLimit )
	
	check( 'LrmOzPurgeAgent', 'uuid', '?binaryStrategy', '?boolean' )

	local indexStore = self._indexStore
	local docStore = self._docStore
	
	local albumDoc = docStore:docValue( albumId )
	
	local spaceRequired = minimumFreeSpace
	if noPurgeLimit then
		spaceRequired = -1
	end	

	-- IMPORTANT!!
	--
	-- Verify offline use is still disabled for the album otherwise we are gonna ddos Oz
	-- as the PurgeAgent will try to remove files locally whereas the BinaryDownloader 
	-- will download again

	local result = albumDoc
		:flatMap( function( docValue )
	
			local useOffline = derefOpt( docValue, '_localOnly', 'useOffline' )
			
			if useOffline == false then
			
				return indexStore:query{
					sql = purgeableAlbumAssetListSql(), 
					quiesceUntilUpToDate = true,
					albumId }
							
			else
			
				return Rx.Observable.just( {} )
				
			end
			
		end )

	
	return self._initializedSubject
		:flatMap( function()
			return self:purgeUntilDone( result, spaceRequired )
		end )
		
end

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

function purgeAgentMethods:stopContinuousPurge()

	check( 'LrmOzPurgeAgent' )

	if self._purgeSub then
		self._purgeSub:dispose()
		self._purgeSub = nil
	end
	
end

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

function purgeAgentMethods:startContinuousPurge( binaryStrategy )

	check( 'LrmOzPurgeAgent', 'binaryStrategy' )

	if self._purgeSub and binaryStrategy == self._binaryStrategy then
		-- if we're already running and the binary strategy hasn't changed, we're done
		return
	end

	if self._purgeSub then
		self:stopContinuousPurge()
	end

	setBinaryStrategy( self, binaryStrategy )

	self.private._purgeContinuously = purgeIsEnabled( self )
		:flatMapLatest( function( enabled )

			local spaceNeeded = self.private._getDiskSpaceNeeded( self )

			if not enabled then
				return Rx.Observable.just( 0, spaceNeeded )
			end

			return self._purgeUserUpdated
				:flatMap( function( potentialVictims )

					return purgeVictims( self, potentialVictims, spaceNeeded )

				end )
		end )
		:replay( nil, 1 )
		:refCount()

	self._purgeSub = self.private._purgeContinuously:subscribe()

	return self.private._purgeContinuously

end

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

function purgeAgentMethods:maybePurgeExternalDevelopSettingsFile( externalDevelopSettingsFilename )
	
	check( 'LrmOzPurgeAgent', '+string' )
	
	return maybePurgeXmpDevelopFiles( self._wfCatalog, { externalDevelopSettingsFilename } )
				
end

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

function purgeAgentMethods:lockAssetForPurge( assetId )
	
	check( 'LrmOzPurgeAgent', 'uuid' )
	
	local assetIdRefCount = self._lockedAssets[ assetId ] or 0
	self._lockedAssets[ assetId ] = assetIdRefCount + 1
	
	if #self._lockedAssets > batchSize / 2 then
		-- If batchSize assets are locked, the locked assets might block other purges from being processed.
		-- Also, there should normally only be one asset locked, so add a warning if there are more than half 
		-- the batchSize
		logger:warnf( "lockAssetForPurge: %d assets are locked", #self._lockedAssets )
	end
	
	logger:debugf( "lockAssetForPurge: self._lockedAssets= %s", self._lockedAssets )
end

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

function purgeAgentMethods:unlockAssetForPurge( assetId )
	
	check( 'LrmOzPurgeAgent', 'uuid' )
	
	local assetIdRefCount = self._lockedAssets[ assetId ]
	if assetIdRefCount then
		
		assetIdRefCount = assetIdRefCount - 1
		self._lockedAssets[ assetId ] = ( assetIdRefCount > 0 ) and assetIdRefCount or nil
	end
	logger:debugf( "unlockAssetForPurge: self._lockedAssets= %s", self._lockedAssets )
end

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


return Rx.Class 'LrmOzPurgeAgent'( purgeAgentMethods )
