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

NmbOfflineAssetsTask.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2018 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains 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 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 Systems Incorporated.

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

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

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFRxContinuousTask = require 'WFRxContinuousTask'

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

	-- Lightroom CC
local NmbOfflineAssets = require 'NmbOfflineAssets'

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

local check = assert( WFArguments.check )

local logger = WFLogger 'NmbOfflineAssetsTask'

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

local NmbOfflineAssetsTaskMethods = {
		
	_wfCatalog = false,
		-- (NmbCatalog)
	
	_continuousOfflineAssetsTask = false,
		-- (WFRxContinuousTask) A continuous task to handle toggling the offline
		-- status of assets.

	_debugMode = false,
		-- (boolean) If true, removes the task delays. Used by unit tests.

	_enablingInProgressIds = false,
		-- (table) A table used to prevent duplicate processing of offline assets.
		-- The table contains asset ids that are currently being enabled for 
		-- offline use.
		-- key: asset id
		-- value: true
		
	_disablingInProgressIds = false,
		-- (table) A table used to prevent duplicate processing of offline assets.
		-- The table contains asset ids that are currently being disabled from 
		-- offline use.
		-- key: asset id
		-- value: true
		
	_processedAssetIdStream = false,
		-- (ReplaySubject) A stream that fires with asset ids as assets are processed.
		-- Used by unit tests.
		
	WFCloseSupport.addCloseSupportMethods,
	
}

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

local function addToRecentlyProcessed( self, enableOfflineUse, assetIds )

	if enableOfflineUse then

		for _, assetId in ipairs( assetIds ) do
			self._enablingInProgressIds[ assetId ] = true
		end
	
	else
	
		for _, assetId in ipairs( assetIds ) do
			self._disablingInProgressIds[ assetId ] = true
		end
	
	end
	
end

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

local function removeFromRecentlyProcessed( self, enableOfflineUse, assetIds )

	if enableOfflineUse then

		for _, assetId in ipairs( assetIds ) do
			self._enablingInProgressIds[ assetId ] = nil
			self._processedAssetIdStream:onNext( { assetId, true } )
		end
	
	else
	
		for _, assetId in ipairs( assetIds ) do
			self._disablingInProgressIds[ assetId ] = nil
			self._processedAssetIdStream:onNext( { assetId, false } )
		end
	
	end
	
end

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

local function clearAllInProgress( self )

	self._enablingInProgressIds = {}
	self._disablingInProgressIds = {}
	
end

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

local function isProcessing( self, enableOfflineUse, assetId )

	if enableOfflineUse then
	
		return self._enablingInProgressIds[ assetId ]
	
	else
	
		return self._disablingInProgressIds[ assetId ]
	
	end

end
 
--------------------------------------------------------------------------------

local function runQueryAndDoWork( self, enableOfflineUse, queryTbl, workFn )

	local indexStore = self._wfCatalog:indexStore()

	local readyForNextBatchOfWork = Rx.ReplaySubject( 1 )
	readyForNextBatchOfWork:onNext( true )

 	-- SPAM [jandrew 2018-06-26]. The following function runs an index store query
	-- that fetches about ~250 assets to be processed. If there are no assets to process
	-- the query is left open until there is work to do. If there _are_ assets to process,
	-- the query is shutdown (via take(1)) and the assets are processed which invovles updating
	-- their doc values. Once the assets are updated, this function is called again to process
	-- more assets. Given the nature of this task, we don't need to worry about processing
	-- the same asset more than once.
	
	-- This design throttles the workload to the doc store, which makes for a better user experience.
	
	local function processBatch()
	
		return indexStore:query( queryTbl )
		
			:filter( function( rows )
			
				return #rows > 0
			
			end )
			
			:take( 1 )
			
			:flatMap( function( rows )
					
				self._continuousOfflineAssetsTask:taskDidWork( true )
					
				local assetIds = {}
				
				for _, row in ipairs( rows ) do
				
					local assetId = row[ 1 ]
					
					if not isProcessing( self, enableOfflineUse, assetId ) then 
						table.insert( assetIds, assetId )
					end
					
				end
			
				local doWork = Observable.just( true )
			
				if #assetIds > 0 then
				
					if enableOfflineUse then
						logger:infof( "Now marking %d assets for offline use", #assetIds )
					else
						logger:infof( "Now disabling %d assets from offline use", #assetIds )
					end
				
					addToRecentlyProcessed( self, enableOfflineUse, assetIds )
				
					doWork = workFn( self._wfCatalog, assetIds )
					
						:concat( Observable.just( true ) )
							-- SPAM [jandrew 2018-06-19]: This concat was added to make sure that the
							-- `flatMap` in the `processAssets` fn is hit.
							
						:tap( function()
						
							removeFromRecentlyProcessed( self, enableOfflineUse, assetIds )

						end )
					
				end

				return doWork
					
			end )
			
			:tap( function()
				readyForNextBatchOfWork:onNext( true )
			end )
		
	end
	
	return readyForNextBatchOfWork
		:flatMap( function()
			return processBatch()
		end )

end

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

--[[!

	\brief Returns a query that selects all of the assets that need to be marked
	for offline use because they are within an album that is marked for offline
	use. Assets already marked for offline use are ignored.

--]]

local function getAssetsNeedingOfflineUseEnabledSql()

	return [[
		SELECT DISTINCT album_asset_v2.assetId
			FROM album_asset_v2
			INNER JOIN assets USING ( assetId )
			WHERE assets.isMarkedForOffline = 0
			AND albumId IN (
				SELECT m.albumId
				FROM albums as m
				WHERE m.isMarkedForOffline = 1 )
			LIMIT 250 ]]
end

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

--[[!

	\brief Returns a query that selects all of the assets that should no longer
	be marked for offline use because they are no longer within an album that 
	is marked for offline use.

--]]

local function getAssetsNeedingOfflineUseDisabledSql()

	return [[ 
		SELECT DISTINCT assetId
			FROM assets
			WHERE isMarkedForOffline = 1
		EXCEPT
		SELECT DISTINCT album_asset_v2.assetId
			FROM album_asset_v2
			INNER JOIN assets USING( assetId )
			WHERE assets.isMarkedForOffline = 1
			AND albumId IN(
				SELECT m.albumId
				FROM albums as m
				WHERE m.isMarkedForOffline = 1 ) 
		LIMIT 250 ]]

end

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

local function processAssets( self )

	local enableOfflineUseWorker = runQueryAndDoWork( 
			self,
			true,
			{
				sql = getAssetsNeedingOfflineUseEnabledSql(),
				quiesceUntilUpToDate = true,
				calm = self._debugMode and 0.1 or 5,
			}, 
			NmbOfflineAssets.enableAssetsOfflineUse )
			
	local disableOfflineUseWorker = runQueryAndDoWork( 
			self,
			false,
			{
				sql = getAssetsNeedingOfflineUseDisabledSql(),
				quiesceUntilUpToDate = true,
				calm = self._debugMode and 0.1 or 10,
			},
			NmbOfflineAssets.disableAssetsOfflineUse )
		
	return Observable.combineLatest( enableOfflineUseWorker, disableOfflineUseWorker )
	
		:flatMap( function()
			return self._continuousOfflineAssetsTask:throwIfNotRunning()
		end )

		:catch( function( errorMessage )
			return self._continuousOfflineAssetsTask:throwIfNotStopped( errorMessage )
		end )
		
		:finally( function() clearAllInProgress( self ) end )

end

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

-- @mark NmbOfflineAssetsTask

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

--[[!

	\brief NmbOfflineAssetsTask is used to update the marked-for-offline-use status
		of assets. This status can change when an asset is added or removed from an
		album -- via sync or manually by the user -- and can also change when an 
		existing parent album's marked-for-offline-use status changes.
			
--]] 

function NmbOfflineAssetsTaskMethods:__init( wfCatalog, optArgs )
	
	check( 'NmbOfflineAssetsTask', 'WFCatalog', '?table' )
	
	optArgs = optArgs or {}
	
	self._wfCatalog = wfCatalog
	
	self._debugMode = optArgs.debugMode
	
	self._enablingInProgressIds = {}
	
	self._disablingInProgressIds = {}
	
	self._processedAssetIdStream = Rx.ReplaySubject( 1 )
	
	logger:infof( "Spinning up NmbOfflineAssetsTask" )
	
	local continuousOfflineAssetsTask = WFRxContinuousTask( "NmbOfflineAssetsTask", 
		function() 
			return processAssets( self ) 
		end, 
		{
			progressiveBackoff = false,
			errorDelay = 5,
			maxDelay = 10,
		} )
	
	self:setOwned {
		_continuousOfflineAssetsTask = continuousOfflineAssetsTask,
	}

	self._continuousOfflineAssetsTask:startContinuousTask()

end

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

function NmbOfflineAssetsTaskMethods:close()
	
	self._continuousOfflineAssetsTask:stopContinuousTask()
	
	local processedAssetIdStream = self._processedAssetIdStream
	self._processedAssetIdStream = nil
	
	if processedAssetIdStream then
		processedAssetIdStream:dispose()
	end
	
end

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

function NmbOfflineAssetsTaskMethods:getProcessedAssetIdStream()
	
	return self._processedAssetIdStream
	
end

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

return Rx.Class( 'NmbOfflineAssetsTask', NmbOfflineAssetsTaskMethods )
