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

PixelProviderFileCachePurgeWorker.lua

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

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

--------------------------------------------------------------------------------]]
	
	-- wf.core
local KSTableUtils = require 'KSTableUtils'
local WFCloseSupport = require "WFCloseSupport"
local WFFunctionUtils = require 'WFFunctionUtils'

	-- WF.core.lua.threading
local WFWorkerThread = require 'WFWorkerThread'
local WFRequestDispatcher = require 'WFRequestDispatcher'

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'

	-- WF.oz_client.managed_catalog
local WFConfiguration = require 'WFConfiguration'
local NmbLocalRenditionDigest = require 'NmbLocalRenditionDigest'
local WFCatPixelPathGenerator = require 'WFCatPixelPathGenerator'

	-- Pixel provider
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'

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

	-- Rx
local Rx = require "Rx"

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

local logger = require 'WFLogger'( "PixelProviderFileCachePurgeWorker" )

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

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt,
										"KSTableUtils.derefOpt does not exist" )

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

local WFFunctionUtils_reduce = WFFunctionUtils.reduce
local WFFunctionUtils_filter = WFFunctionUtils.filter

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

local function blockWithShouldContinue()

	return coroutine.yield() ~= "quit" 

end

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

local function swap( arr, i, j )

	if i == j then

		return arr

	end

	local v = arr[ i ]

	arr[ i ] = arr[ j ]

	arr[ j ] = v

	return arr

end

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

local function shuffle( arr )

	for _ = 1, #arr do

		arr = swap( arr, 1, math.random( #arr ) )
		
	end

	return arr

end

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

local function getNextScanFolder( self, prevScanFolderOffset )

	-- now see if we've request a full scan
	if not self._scanAllFoldersData or 
			not next( self._scanAllFoldersData.folderPathsAndTypeNeedingScan ) then

		-- no full scan
		return

	end

	-- we've requested a full scan, find the next folder, skipping any 
	-- folders that we've already scanned for deletion
	
	local currScanFolderOffset = ( prevScanFolderOffset or 0 ) + 1

	local scanFolder, scanFolderType

	repeat 

		scanFolder = self._scanAllFoldersData.sequence[ currScanFolderOffset ]
		scanFolderType = self._scanAllFoldersData.folderPathsAndTypeNeedingScan[ scanFolder ]
		
		if not scanFolder then

			-- we've finished scanning all of the folders
			return

		end

		if not scanFolderType then

			-- we no longer need to scan this folder, skip it
			scanFolder = nil

			currScanFolderOffset = currScanFolderOffset + 1

		end

	until scanFolder

	return scanFolder, scanFolderType, currScanFolderOffset
	
end

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

local function extractAssetIdDigestPair( docValue )

	-- return `assetId` if the record is deleted or not present
	--
	-- return `assetId, digest` if we have a record and a digest
	--
	-- return `nil` otherwise

	local assetId = KSTableUtils_derefOpt( docValue, "_id" )

	if not assetId then

		return 

	end

	if KSTableUtils_derefOpt( docValue, "_deleted" ) or 
			KSTableUtils_derefOpt( docValue, "_empty" ) then

		return assetId
		
	end

	local digest = NmbLocalRenditionDigest.maybeComputeFromDocValue( 
					docValue )

	return assetId, digest

end

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

local function isNonNil( v )

	return v ~= nil

end

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

local function isNonEmpty( v )
	
	return type( v ) == 'table' and next( v )
	
end

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

local function getObsoleteRenditions( cacheFolderScanResults, assetId,
										currDigest )

	if not assetId then

		return {}

	end

	if not currDigest then

		-- all of the results are obsolete
		return cacheFolderScanResults[ assetId ]

	end

	if not KSTableUtils_derefOpt( 
			cacheFolderScanResults, assetId, currDigest ) then

		-- we don't have any renditions for this digest, hold off on deleting
		return {}

	end

	local renditionsForAssetId = cacheFolderScanResults[ assetId ] or {}

	local obsoleteAssetIdRenditions = {}

	for otherDigest, otherDigestVals in pairs(  renditionsForAssetId ) do

		if otherDigest ~= currDigest then

			obsoleteAssetIdRenditions[ otherDigest ] = otherDigestVals

		end

	end

	return obsoleteAssetIdRenditions


end

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

local function extractKeys( t )

	local keys = {}

	for k in pairs( t ) do

		keys[ #keys + 1 ] = k

	end

	return keys

end

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

local function addToChunkedArray( chunks, val, chunkSize )

	local currChunk = chunks[ #chunks ]

	local currChunkSize = #currChunk
	
	if currChunkSize > 0 and currChunkSize  % chunkSize == 0 then

		currChunk = {}

		chunks[ #chunks + 1 ] = currChunk

	end
		
	currChunk[ #currChunk + 1 ] = val

	return chunks

end

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

local function chunkedArray( arr, chunkSize )

	return WFFunctionUtils_reduce( { {} }, addToChunkedArray, arr, chunkSize )

end

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

local function makeObsoleteAssetIdRenditionsStream( docStore, cacheFolderScanResults )

	return Rx.Observable.create( 

		function( observer )

			local assetIds = extractKeys( cacheFolderScanResults or {} )

			local bulkDocValueStream =
					WFRxThrottledBatchedBulkDocValue( "PixelProviderFileCachePurgeWorker", docStore, assetIds, 50, 10 )

			-- make a stream of assetIds that have renditions

			-- for those assetIds, get a docStore stream of the latest assetId,
			-- digest values
			local currAssetIdDigestPairStream =
					bulkDocValueStream
						:flatMap( Rx.Observable.from )
						:map( extractAssetIdDigestPair )
						:filter( isNonNil )

			-- Now build a stream obsolete renditions for the assetId
			local obsoleteAssetIdRenditionsStream = 
					currAssetIdDigestPairStream
						:map(
							function( assetId, currDigest )
			
								return assetId,
										getObsoleteRenditions(
											cacheFolderScanResults, assetId,
											currDigest )

							end )

			-- subscribe to this stream
			return obsoleteAssetIdRenditionsStream:subscribe( observer )

		end )

end

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

local function purgedRenditionsFilesStream( self, wfCatalog, scanFolder )

	-- make a stream of renditions for that folder
	local cacheFolderScanResultStream =
		self._purgeIOWorker:request( 'scanRenditionFolderNamed', 
										scanFolder )

	-- make a stream of obsolete assetId renditions for that folder
	local obsoleteAssetIdRenditionsStreamOfStreams =
		Rx.Observable.combineLatest(
			makeObsoleteAssetIdRenditionsStream,
				Rx.Observable.just( wfCatalog:docStore() ), 
				cacheFolderScanResultStream  )

	-- make a stream of assetId rendition purge results for folder
	return obsoleteAssetIdRenditionsStreamOfStreams
		:switch()
		:flatMap(
			function( assetId, obsoleteAssetIdRenditions )

				if not next( obsoleteAssetIdRenditions or {} ) then

					return Rx.Observable.just( {} )

				end

				return self._purgeIOWorker:request( 
							'deleteRenditions', 
								{ [ assetId ] =
									obsoleteAssetIdRenditions } )

			end )

end

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

local function isSettingsSha256NoLongerInUse( sha256, sha256SettingsInUse )
	if not sha256SettingsInUse[ sha256 ] then
		-- return the sha256 value if the sha256 is *not* in use
		return sha256
	end
end

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

local function makeObsoleteSettingsStream( indexStore, cacheFolderScanResults )

	return Rx.Observable.create( 

		function( observer )

			local chunkedFolderSettingsSha256s = 
					chunkedArray( extractKeys( cacheFolderScanResults ), 50 )

			-- make a stream of assetIds that have renditions

			-- for those assetIds, get a docStore stream of the latest assetId,
			-- digest values
			local currSettingsForPossiblePurgeStream =
					Rx.Observable.from( chunkedFolderSettingsSha256s )
						:filter( isNonNil )
			
			local allSha256SettingsInUseStream = indexStore:queryOnce {
				sql = [[SELECT sha256 from externalDevelopSettings]],
				quiesceUntilUpToDate = true,
				singleColumnQuery = true
			}
			:map( function( rows )
				local sha256Map = {}
				for _, sha256 in ipairs( rows ) do
					sha256Map[ sha256 ] = true
				end
				return sha256Map
			end )
			
			-- Now build a stream obsolete renditions for the assetId
			local obsoleteSettingsStream = allSha256SettingsInUseStream
				:flatMapLatest( function( allSha256SettingsInUse )
					return currSettingsForPossiblePurgeStream
						:map( function( settingsForPossiblePurge )
							return settingsForPossiblePurge, allSha256SettingsInUse
						end )
				end )
				:map( function( settingsSha256ForPossiblePurge, allSha256SettingsInUse )
					
					return WFFunctionUtils_filter( isSettingsSha256NoLongerInUse, settingsSha256ForPossiblePurge, allSha256SettingsInUse )

				end )
				:filter( isNonEmpty )

			-- subscribe to this stream
			return obsoleteSettingsStream:subscribe( observer )

		end )
end

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

local function purgedSettingsFilesStream( self, wfCatalog )

	local cacheFolderScanResultStream = self._purgeIOWorker:request( 'scanSettingsFolder', self._minSettingsXmpAgeInSeconds )
	
	local obsoleteSettingsStreamOfStreams = 
		Rx.Observable.combineLatest( 
			makeObsoleteSettingsStream,
				Rx.Observable.just( wfCatalog:indexStore() ),
				cacheFolderScanResultStream )
	
	return obsoleteSettingsStreamOfStreams
		:switch()
		:flatMap(
				function( obsoleteSha256Settings )
	
					if not next( obsoleteSha256Settings or {} ) then
	
						return Rx.Observable.just( {} )
	
					end
	
					return self._purgeIOWorker:request( 
								'deleteSettings', obsoleteSha256Settings )
	
				end )

end

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

local function resumeCoroutineUnlessRunning( theCoroutine, ... )

	local currStatus = coroutine.status( theCoroutine )
	
	if currStatus == "dead" then


		return false, "tried to resume dead couroutine"

	end

	if currStatus == "running" then

		return false, "tried to resume couroutine from itself"

	end

	return coroutine.resume( theCoroutine, ... )

end

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

local function cacheScanLoopImp( self, wfCatalog )

	-- this is a coroutine "thread". It runs separate from the methods in self
	-- and also separate from the Rx subscription callbacks. Therefore it can
	-- block without causing other work to stop
	--
	-- We take advantage of that to implement the following:
	--
	-- 	1) iterate all root level folders of the rendition cache in random
	-- 	   order
	-- 	2) For each folder, find purgeable renditions
	-- 	3) Purge them
	-- 	4) Block until the above completes, the start on the next folder

	local scanFolder, scanFolderType, currScanFolderOffset

	local shouldKeepScanning = true

	while shouldKeepScanning do

		-- get a new root level rendition folder
		scanFolder, scanFolderType, currScanFolderOffset =
			getNextScanFolder( self, currScanFolderOffset )

		if not scanFolder then

			return "done"

		end
		
		local purgedFilesResultsStream
		
		if scanFolderType == 'settings' then
			-- external develop settings
			purgedFilesResultsStream = purgedSettingsFilesStream( self, wfCatalog )
		
		else 
			-- renditions
			purgedFilesResultsStream = purgedRenditionsFilesStream( self, wfCatalog, scanFolder )
		end
	
		-- finishedPurgingFolder prevents us from blocking just in case
		-- the implementations become synchronous
		local finishedPurgingFolder = false
	
		-- subscribe to purgedFilesResultsSubscription, which
		-- is what causes the streams to start working
		local purgedFilesResultsSubscription =
				purgedFilesResultsStream 			
					:subscribe(
	
						function( deleteResults )
	
							if logger.will_info then
	
								logger:infof( "Finished trying to delete %s %s from folder '%s'",
													#deleteResults, scanFolderType, scanFolder )
	
							end
	
						end,
	
						function( errStr )
	
							logger:errorf( "Error trying to purge folder '%s': %s", scanFolder, errStr )
	
							finishedPurgingFolder = true
	
							resumeCoroutineUnlessRunning(
								self._cacheScanLoopCoroutine )
	
						end,
	
						function()
	
							logger:infof( "Finished purging folder '%s'", scanFolder )
	
							finishedPurgingFolder = true
	
							resumeCoroutineUnlessRunning(
								self._cacheScanLoopCoroutine )
	
						end )
	
		-- now that we are purging, we block this coroutine loop
	
		shouldKeepScanning = finishedPurgingFolder or blockWithShouldContinue()
	
		-- once we unblock, we've either finished the subscription or
		-- self is closing. In either case, it's time to dispose the
		-- subscription
		purgedFilesResultsSubscription:dispose()
	
	end

	return "done"

end

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

local function cacheScanLoop( self, wfCatalog )

	local result = cacheScanLoopImp( self, wfCatalog )

	self._scanAllFoldersData = nil

	self._scanStatusStream:onNext( result )

	return result

end

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

local function createCacheScanCoroutine( self, wfCatalog )

	local cacheScanLoopCoroutine = coroutine.create( cacheScanLoop )

	-- pass initial arguments to runLoop
	coroutine.resume( cacheScanLoopCoroutine, self, wfCatalog )

	return cacheScanLoopCoroutine

end

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

local function maybeStartCacheScan( self )

	if type( self._cacheScanLoopCoroutine ) ~= "thread" or
			coroutine.status( self._cacheScanLoopCoroutine ) == "dead" then

		self._cacheScanLoopCoroutine = 
				createCacheScanCoroutine( self, self._wfCatalog )

	end

end

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

local PixelProviderFileCachePurgeWorker = {
	
	_wfCatalog = false,

	_purgeIOWorker = false,
		-- (WFWorkerThread)
		
	_deletedAssetIdsSubscription = false,
	
	_deletedStatusStream = false,
	
	_scanAllFoldersData = false,

	_cacheScanLoopCoroutine = false,
		-- coroutine that scans the rendition cache for stale renditions

	_scanStatusStream = false,
	
	_minSettingsXmpAgeInSeconds = false,
		-- minimum age of settings XMP before we even consider purging them
		
	WFCloseSupport.addCloseSupportMethods,

}

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

local WFCatPixelPathGenerator_renditionFolderNameForAssetId =
		assert( WFCatPixelPathGenerator.renditionFolderNameForAssetId,
					"missing WFCatPixelPathGenerator.renditionFolderNameForAssetId" )

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

local function requiresIndividualPurge( docValue )
	if type ( docValue._localOnly ) == "table" then
		return not docValue._localOnly.deletedLocally
	else
		return true
	end
end

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

function PixelProviderFileCachePurgeWorker:__init( wfCatalog, options )

	logger:info( "PixelProviderFileCachePurgeWorker is created" )
	
	local pixelsPath = wfCatalog:fileSpec():pixelsPath()
	
	self._minSettingsXmpAgeInSeconds = ( options and options.minSettingsXmpAgeInSeconds ) or
		WFConfiguration.PixelProvider.SettingsXmpCachePurgeOlderThan or 
									( 2 * 24 * 60 * 60 ) -- default is 2 days, to be very safe
	
	self:setOwned {

		_wfCatalog = wfCatalog,

		_purgeIOWorker = WFWorkerThread( 'WFCatPixelFileIOWorker.lua',
											pixelsPath ),
	
	}

	local purgeWorkerChangeCounterLocalDocId = 'PurgeWorker-changeCounter'

	local docStore = wfCatalog:docStore()

	self._deletedAssetIdsSubscription = docStore:deletedDocsSinceNamedChangeCounter( 
			purgeWorkerChangeCounterLocalDocId, 
			{
				_localOnly = {
					deletedLocally = true
				},
				_id = true,
				type = true
			} )

		:map( function( docValues, continueFrom )
		
			local assetIds = {}
			
			for _, docValue in ipairs( docValues ) do
			
				-- deletes from Oz do not have type info, so we have to choose
				-- between doing scans for possible non-assets or waiting for the
				-- next full rendition scan to clean up missed deletes
				
				if docValue 
					and ( docValue.type == "asset" or docValue.type == nil )
					and requiresIndividualPurge( docValue ) then
					
					assetIds[ #assetIds + 1 ] = docValue._id
					
				end
				
			end
				
			return assetIds, continueFrom
				
		end )

		:flatMap( function( assetIds, continueFrom )
		
			local result = Rx.Observable.just( continueFrom )
			
			if #assetIds == 0 then
			
				return result
			
			end
			
			return Rx.Observable.from( assetIds )
			
				:flatMap( function( assetId )
			
					return self._purgeIOWorker:request( 'deleteAssetRenditions', assetId )
					
						:concat( Rx.Observable.just( true ) )
							-- Concat a simple observable that fires so we can map the result.
							
						:map( function()
						
							self._deletedStatusStream:onNext( assetId )
							
							return continueFrom
							
						end )
				
				end )
				
				:takeLast( 1 )
			
		end )
		
		:flatMap( function( continueFrom )
		
			return docStore:updateDeletedDocsSinceNamedChangeCounter( 
					purgeWorkerChangeCounterLocalDocId, continueFrom )
					
		end )
		
		:subscribe()
		
	self._deletedStatusStream = Rx.Subject()
	
	self._scanStatusStream = Rx.Subject()

end

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

local function maybeQuitCoroutine( theCoroutine )

	if type( theCoroutine ) ~= "thread" then

		return

	end

	local currStatus = coroutine.status( theCoroutine )
	
	if currStatus == "dead" then

		return

	end

	local ok, maybeLoopExitStatus = resumeCoroutineUnlessRunning( 
										theCoroutine, "quit" )

	if not ok then

		logger:errorf( "failed to quit the runLoop: %s", 
							maybeLoopExitStatus )

		return 

	end

	local finalStatus = coroutine.status( theCoroutine )

	if maybeLoopExitStatus ~= "done" or finalStatus ~= "dead" then

		logger:errorf( "run loop did not exit cleanly. Loop exit status: %s. Coroutine status: %s",
							maybeLoopExitStatus, finalStatus )

	end

end

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

function PixelProviderFileCachePurgeWorker:close()

	logger:info( "PixelProviderFileCachePurgeWorker is closing" )
	
	self._deletedAssetIdsSubscription:dispose()

	self._deletedStatusStream:dispose()
	
	self._scanStatusStream:dispose()

	maybeQuitCoroutine( self._cacheScanLoopCoroutine )

end

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

function PixelProviderFileCachePurgeWorker:purgeResultsStream()

	return self._purgeIOWorker:request( 'purgeResultsStream' )

end

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

function PixelProviderFileCachePurgeWorker:scanAndPurge()

	if self._scanAllFoldersData then

		-- already scanning
		return

	end

	-- start with rendition folders
	local allScannableFolders = WFCatPixelPathGenerator.assetIdRenditionFolderNames()
	local allScannableFoldersSet = {}
	for _, folder in ipairs( allScannableFolders ) do
		allScannableFoldersSet[ folder ] = 'renditions'
	end
	
	-- add settings folder
	table.insert( allScannableFolders, 'settings' )
	allScannableFoldersSet[ WFCatPixelPathGenerator.settingsXmpFolderName() ] = 'settings'
	
	-- shuffle the folders we're going to scan so we don't always scan in the same order
	local allScannableFoldersInRandomOrder =
			shuffle( allScannableFolders )
	
	self._scanAllFoldersData = {

		sequence = allScannableFoldersInRandomOrder,
		
		folderPathsAndTypeNeedingScan = allScannableFoldersSet,

	}

	maybeStartCacheScan( self )

end

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

function PixelProviderFileCachePurgeWorker:purgeRenditionsForDeletedAssets(
		assetIds )
	
	local assetIdSet = KSTableUtils.mapArrayToSet( assetIds )
	local folderSet = { }

	return Rx.Observable.pairs(assetIdSet)
		:map( WFCatPixelPathGenerator_renditionFolderNameForAssetId )
		:filter( function( folder )
			if folderSet[ folder ] then
				return false
			else
				folderSet[ folder ] = true -- once each
				return true
			end
		end)
		:map( function( folder )
			return self._purgeIOWorker:request( 'scanRenditionFolderNamed', folder )
		end )
		:flatMerge( 2 )
		:map( function( renditionInfos )
			for assetId in pairs( renditionInfos ) do
				if not assetIdSet[ assetId ] then
					-- clear this item out, it is not in our deleted set
					renditionInfos[ assetId ] = nil
				end
			end

			-- 3. call deleteRenditions for the renditionInfos
			-- that matched our deleted assetId set
			return self._purgeIOWorker:request( 'deleteRenditions', renditionInfos )
		end )
		:flatMerge( 2 )
		:takeLast( 0 )

end

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

function PixelProviderFileCachePurgeWorker:deletedStatusStream()

	return self._deletedStatusStream

end

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

function PixelProviderFileCachePurgeWorker:scanStatusStream()

	return self._scanStatusStream

end

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

PixelProviderFileCachePurgeWorker.class = {}

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

function PixelProviderFileCachePurgeWorker.class.initMessageProcessor( 
												Class, 
												encodedPackedWFCatalogArgs )

	local packedWFCatalogArgs = KSJsonDecoder.decodeWebSafeJson(
								encodedPackedWFCatalogArgs )

	local wfCatalog = PixelProviderSharedTaskUtils.makeWFCatalog( unpack( packedWFCatalogArgs ) )

	local pixelProviderFileCachePurgeWorker = Class( wfCatalog, packedWFCatalogArgs )
	
	WFCloseSupport.mustCloseAtThreadTermination( 
					pixelProviderFileCachePurgeWorker )

	WFRequestDispatcher.onMessageDo {

		message = 'purgeResultsStream',

		target = pixelProviderFileCachePurgeWorker,

	}

	WFRequestDispatcher.onMessageDo {

		message = 'scanAndPurge',

		target = pixelProviderFileCachePurgeWorker,

	}
	
	WFRequestDispatcher.onMessageDo {

		message = 'purgeRenditionsForDeletedAssets',

		target = pixelProviderFileCachePurgeWorker,

	}

	WFRequestDispatcher.onMessageDo {

		message = 'deletedStatusStream',

		target = pixelProviderFileCachePurgeWorker,

	}

	WFRequestDispatcher.onMessageDo {

		message = 'scanStatusStream',

		target = pixelProviderFileCachePurgeWorker,

	}

end


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

return Rx.Class( 'PixelProviderFileCachePurgeWorker', 
					PixelProviderFileCachePurgeWorker )
