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

PixelPathProviderSchedulerPrivateMethods.lua

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

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

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law. Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

	-- wf.core
local WFArguments = require 'WFArguments'
local WFArguments_check = assert( WFArguments.check,
										"missing WFArguments.check" )
local WFFiniteStateMachine = require "WFFiniteStateMachine"
local WFFunctionUtils = require "WFFunctionUtils"
local WFPerformanceUtils = require "WFPerformanceUtils"
local WFQueueing = require "WFQueueing"

	-- scheduler helpers
local PixelPathProviderRequestTypes = require 'PixelPathProviderRequestTypes'
local PixelPathProviderSchedulerDiagnosticLog =
		require "PixelPathProviderSchedulerDiagnosticLog"
local PixelPathProviderStateMachine = require "PixelPathProviderStateMachine"

	-- Rx
local Rx = require "Rx"

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

local WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription =
		assert( WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithStaticDescription,
				 "No WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithStaticDescription" )

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

local WFFunctionUtils_map = assert( WFFunctionUtils.map,
			"WFFunctionUtils.map does not exist" )

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

local WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

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

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

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

local function forwardToPriorityQueue( funcName )

	return function( scheduler, ... )

		return scheduler._requestPriorityQueue[ funcName ](
					scheduler._requestPriorityQueue, ... )

	end

end

--==============================================================================
-- run loop related
--==============================================================================

local function maybeGetRunLoopCoroutineThreadStatus( scheduler )

	return scheduler._runLoopCoroutineThread and 
			coroutine.status( scheduler._runLoopCoroutineThread ) or false

end

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

local function shouldResumeRunLoop( scheduler )

	if scheduler._batchScheduleRequests and scheduler._batchScheduleRequests > 0 then

		return false

	end

	local maybeRunLoopCoroutineThreadStatus = 
			maybeGetRunLoopCoroutineThreadStatus( scheduler )

	if not maybeRunLoopCoroutineThreadStatus then

		logger:error( "Tried to schedule requests without an active runloop" )

		return false

	end

	if maybeRunLoopCoroutineThreadStatus ~= "suspended" then

		logger:debugf( "tried to resume _runLoopCoroutineThread when it wasn't suspended. status = %s", 
						maybeRunLoopCoroutineThreadStatus )

		return false

	end

	return true

end

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

local function resumeRunLoopCoroutine( scheduler )

	return coroutine.resume( scheduler._runLoopCoroutineThread )

end

--==============================================================================
-- sending data to observables
--==============================================================================

local function isRunLoopCoroutineThreadActive( scheduler )

	return maybeGetRunLoopCoroutineThreadStatus( scheduler ) ~= "suspended"

end

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

local function sendDataFromSchedulerThread( scheduler, stream, backlogPush, 
												backlogPop, ... )

	local nArgs = select( "#", ... )

	if nArgs > 0 then

		backlogPush( { n = nArgs, ... } )

	end

	local sentPerPixStream

	if stream == scheduler._anyPixStreamIdPixelPathStream and
			scheduler._DEBUG_requestStatsStreamSubscriberCount and
			scheduler._DEBUG_requestStatsStreamSubscriberCount > 0 then

		sentPerPixStream = rawget( scheduler, "_DEBUG_pixelPathsSentPerPixStream" ) or {}

		rawset( scheduler, "_DEBUG_pixelPathsSentPerPixStream", sentPerPixStream )

	end

	while true do

		-- don't send data while the runLoop coroutine is active
		if isRunLoopCoroutineThreadActive( scheduler ) then

			break

		end

		local backlogItem = backlogPop()

		if not backlogItem then

			break

		end

		stream:onNext( unpack( backlogItem, 1, backlogItem.n ) )

		if sentPerPixStream then

			local pixStreamId = backlogItem[ 1 ]

			sentPerPixStream[ pixStreamId ] = 
				( sentPerPixStream[ pixStreamId ] or 0 ) + 1

		end

	end

end

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

local function sendToPixelStreamImp( scheduler, ... )

	if not scheduler._pixelPathStreamBacklogQueue then

		scheduler._pixelPathStreamBacklogQueue = {}

		scheduler._pixelPathStreamBacklogQueue.push, 
			scheduler._pixelPathStreamBacklogQueue.pop = 
				WFQueueing.makeQueue()

	end

	sendDataFromSchedulerThread( scheduler, scheduler._anyPixStreamIdPixelPathStream,
									scheduler._pixelPathStreamBacklogQueue.push,
									scheduler._pixelPathStreamBacklogQueue.pop,
									... )

end

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

local function drainPixelStreamBacklog( scheduler )

	sendToPixelStreamImp( scheduler )

end

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

local function drainBacklogs( scheduler )

	-- send any data that accumulated while running

	drainPixelStreamBacklog( scheduler )

	local drainTestDataBacklog = rawget( scheduler, "_drainTestDataBacklog" )

	if drainTestDataBacklog then

		drainTestDataBacklog( scheduler )

	end

end

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

local function runScheduledRequests( scheduler )

	if not shouldResumeRunLoop( scheduler ) then

		return

	end

	-- run the requests on the run loop

	local status, maybeErr = resumeRunLoopCoroutine( scheduler )

	if not status then

		logger:errorf( "_runLoopCoroutineThread failed to resume with: %s", maybeErr )

	end

	drainBacklogs( scheduler )

end

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

local function wrapInScheduleRequests( f )

	return function( scheduler, ... )

		local ret = { f( scheduler, ... ) }

		runScheduledRequests( scheduler )

		return unpack( ret )

	end

end

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

local _benchmarkedBatchFuncs = {}

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

local function wrapFuncInBenchmarking( desc, f, maybeBenchmarkThreshold )

	local benchmarkThreshold = maybeBenchmarkThreshold or 0.2

	_benchmarkedBatchFuncs[ desc ] = _benchmarkedBatchFuncs[ desc ] or {}

	local benchmarkedBatchFuncsByDesc = _benchmarkedBatchFuncs[ desc ]

	local benchmarkedF =
			benchmarkedBatchFuncsByDesc[ f ] or
			WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithStaticDescription( 
				"PixelProviderScheduler", benchmarkThreshold, desc, f )

	benchmarkedBatchFuncsByDesc[ f ] = benchmarkedF

	return benchmarkedF

end

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

local function runSchedulerFuncInTransaction( f, scheduler, ... )

	return scheduler._requestPriorityQueue:runInTransaction( f, scheduler, ... )

end

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

local function wrapSchedulerFunctionInTransaction( funcName, f )

	local benchmarkedF = wrapFuncInBenchmarking( funcName, f )

	return function( scheduler, ... )

		WFArguments_check( 'PixelPathProviderScheduler' )

		return runSchedulerFuncInTransaction( benchmarkedF, scheduler, ... )

	end

end

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

local function runSchedulerFuncInBatchMode( f, scheduler, ... )

	WFArguments_check( "function", "PixelPathProviderScheduler"  )

	scheduler._batchScheduleRequests = ( scheduler._batchScheduleRequests or 0 ) + 1

	local ret = { pcall( runSchedulerFuncInTransaction, f, scheduler, ... ) }

	scheduler._batchScheduleRequests = scheduler._batchScheduleRequests - 1

	if not ret[ 1 ] then

		logger:error( ret[ 2 ] )

		error( ret[ 2 ] )

	end

	runScheduledRequests( scheduler )

	return unpack( ret, 2 )

end

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

local function wrapSchedulerFunctionInBatchScheduleRequests( 
				desc, f, maybeBenchmarkThreshold )

	WFArguments_check( "string", "function", "?number" )

	local benchmarkedF = wrapFuncInBenchmarking( desc, f, maybeBenchmarkThreshold )

	return function( scheduler, ... )

		return runSchedulerFuncInBatchMode( benchmarkedF, scheduler, ... )

	end

end

--==============================================================================
-- private methods used in run loop
--==============================================================================

local PixelPathProviderRequestTypes_priorities =
		PixelPathProviderRequestTypes.priorities

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

local function getNextNRequestsWithState( scheduler, n, ... )

	if select( '#', ... ) == 1 then

		return scheduler._requestPriorityQueue:getNextNRequestsWithState( n, ... )

	end

	local ret

	local requestPriorityQueue = scheduler._requestPriorityQueue

	for priority = 1, #PixelPathProviderRequestTypes_priorities do

		local currRet =
				requestPriorityQueue:getNextNRequestsWithPriorityAndStates( 
													n, priority, ... )

		if #currRet >= n and not ret then

			return currRet

		end

		if not ret then

			ret = currRet

		else

			local offset = #ret

			for i = 1, #currRet do

				if i + offset > n then

					return ret

				end

				ret[ i + offset ] = currRet[ i ]

			end

		end

		if #ret >= n then

			return ret

		end

	end

	return ret

end

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

local function getNextRequestWithState( scheduler, ... )

	return unpack( getNextNRequestsWithState( scheduler, 1, ... )[ 1 ] or {} )

end

--------------------------------------------------------------------------------
-- cache this lookup so we avoid many calls to 
-- WFFiniteStateMachine.getSourceAndNextStatePairsForTransition
--------------------------------------------------------------------------------

local _fsmStatePairsForTransition = {}

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

local function getSourceAndNextStatePairsForTransition( fsm, transition )

	local fsmStatePairsForTransition = _fsmStatePairsForTransition[ fsm ] or {}

	fsmStatePairsForTransition[ transition ] =
		fsmStatePairsForTransition[ transition ] or
		WFFiniteStateMachine.getSourceAndNextStatePairsForTransition(
			fsm, transition )

	_fsmStatePairsForTransition[ fsm ] = fsmStatePairsForTransition

	return fsmStatePairsForTransition[ transition ]

end

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

local function setNextStateForLocalAssetIdWithState( statePair, 
														requestPriorityQueue, 
														localAssetId )
									
	local sourceState, nextState = unpack( statePair )

	return requestPriorityQueue:setNextStateForLocalAssetIdWithState( 
						sourceState, localAssetId, nextState )

end

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

local function maybeApplyTransitionForLocalAssetId( scheduler, localAssetId, 
													transition )

	local statePairs = getSourceAndNextStatePairsForTransition(
							scheduler._stateMachine, transition )

	WFFunctionUtils_map( 
			setNextStateForLocalAssetIdWithState, 
				statePairs, scheduler._requestPriorityQueue, localAssetId )

end

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

local WFFiniteStateMachine_peekNextState = WFFiniteStateMachine.peekNextState

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

local function getNextStateOrErrState( scheduler, state, transition )

	local newState, maybeErr =
			WFFiniteStateMachine_peekNextState( scheduler._stateMachine, state,
													transition )

	if not newState then

		newState = PixelPathProviderStateMachine.SendDataStates.SendPermanentError

		logger:errorf( "getNextState ( %s, %s ) fail. Setting to '%s' because '%s'",
							state, transition, newState, maybeErr )

	end

	return newState

end

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

local function maybeSetNextState( scheduler, pixStreamId, localAssetId, transition )

	WFArguments_check( 'PixelPathProviderScheduler', 'string', 'number', 
							'string' )

	local currState = scheduler._requestPriorityQueue:maybeGetCurrState(
													pixStreamId, localAssetId )

	if not currState then

		logger:infof( "_maybeSetNextState fail for %s,%s: no matching request to apply transition %s to",
							pixStreamId, localAssetId, transition )

		return

	end
	
	local newState =
			WFFiniteStateMachine_peekNextState( scheduler._stateMachine, currState,
													transition )
													
	if not newState then
	
		logger:infof( "_maybeSetNextState fail for %s,%s: no transition from %s for %s",
					pixStreamId, localAssetId, currState, transition )
	
		return
	
	end
	
	logger:infof( "setNextState %s,%s %s + %s -> %s",
					pixStreamId, localAssetId, currState, transition, newState )

	scheduler._requestPriorityQueue:setState( pixStreamId, localAssetId, newState )

	return newState

end

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

local function applyRequestTransitionOnScheduler( request, scheduler, transition )

	WFArguments_check( 'table', 'PixelPathProviderScheduler', 'string' )

	local pixStreamId, localAssetId, _, _, currState = unpack( request )

	if type( pixStreamId ) ~= "string" or type( localAssetId ) ~= "number" or
			type( currState ) ~= "string" then

		logger:error( "applyTransitionToRequest failing due to invalid request" )

		return

	end

	local newState = getNextStateOrErrState( scheduler, currState, transition )

	logger:infof( "setNextState %s,%s %s + %s -> %s",
					pixStreamId, localAssetId, currState, transition, newState )

	scheduler._requestPriorityQueue
		:setState( pixStreamId, localAssetId, newState )

	return newState

end

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

local function applyTransitionToRequest( scheduler, request, transition )

	return applyRequestTransitionOnScheduler( request, scheduler, transition )

end

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

local function applyTransitionAllRequests( scheduler, requests, transition )

	WFArguments_check( 'PixelPathProviderScheduler', 'table', 'string' )

	WFFunctionUtils_map( applyRequestTransitionOnScheduler, 
							requests, scheduler, transition )

end

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

local function applyTransitionToAllWithState( scheduler, currState, transition )

	local newState = getNextStateOrErrState( scheduler, currState, transition )

	scheduler._requestPriorityQueue:changeStateForAllWithState( currState, newState )

end

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

local function runFuncOnEachRequestWithState( 
					scheduler, 
						requestFetchFunc, 
							requestFetchFuncArgs, 
						mapFn, 
							f, 
								... )

	WFArguments_check( 'PixelPathProviderScheduler', 'function', 'table',
						'function', 'function' )

	local nArgs = select( '#', ... )

	local args = { ... }

	return mapFn( 

		function( request ) 

			return f( request, unpack( args, 1, nArgs ) )

		end,

		requestFetchFunc( scheduler, unpack( requestFetchFuncArgs ) ) )

end

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

local getAllPixstreamIdRequests =
		forwardToPriorityQueue( 'getAllPixstreamIdRequests' )

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

local function mapAllRequestsWithPixStreamId( scheduler, pixStreamId, f, ... )

	return WFFunctionUtils_map(
				f, 
					getAllPixstreamIdRequests( scheduler, pixStreamId ), ... )

end

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

local getRequestsWithStatesInPriorityOrder =
		forwardToPriorityQueue( 'getRequestsWithStatesInPriorityOrder' )

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

local function mapAllRequestsWithStates( scheduler, states, f, ... )

	WFArguments_check( 'PixelPathProviderScheduler', 'table', 'function' )

	return runFuncOnEachRequestWithState( 
				scheduler, getRequestsWithStatesInPriorityOrder, states,
				WFFunctionUtils_map, 
				f, ... )

end

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

local function mapAllRequestsWithState( scheduler, state, f, ... )

	return mapAllRequestsWithStates( scheduler, { state }, f, ... )

end

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


local function reduceAllRequestsWithStates( scheduler, initVal, states, f, ... )

	WFArguments_check( 'PixelPathProviderScheduler', '_', 'table', 'function' )

	local requests = getRequestsWithStatesInPriorityOrder( 
						scheduler, unpack( states ) )

	return WFFunctionUtils_reduce( initVal, f, requests, ... )

end

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

local function sendRenditionPathToPixelStream( 
					scheduler, pixStreamId, localAssetId, pixelPath, renderDigest,
					renderSize, croppedLongSide )

	sendToPixelStreamImp( scheduler, pixStreamId, localAssetId, true, pixelPath,
							renderDigest, renderSize, croppedLongSide )

end

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

local function sendLocalBinaryPathToPixelStream( 
					scheduler, pixStreamId, localAssetId, localBinaryPath,
					ozBinaryType )
	
	sendToPixelStreamImp( scheduler, pixStreamId, localAssetId, true,
							localBinaryPath, ozBinaryType )

end

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

local function sendErrorToPixelStream( 
					scheduler, pixStreamId, localAssetId, errStr )

	sendToPixelStreamImp( scheduler, pixStreamId, localAssetId, false, errStr )

end

--==============================================================================
-- private test methods
--==============================================================================

local function suspendRunLoop( scheduler )

	scheduler._batchScheduleRequests = ( scheduler._batchScheduleRequests or 0 ) + 1

	return scheduler

end

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

local function resumeRunLoop( scheduler )

	scheduler._batchScheduleRequests = ( scheduler._batchScheduleRequests or 0 ) - 1

	return scheduler

end

--------------------------------------------------------------------------------
-- use addRunLoopTaskToAllowlist to control which part of 
-- PixelPathProviderSchedulerMainLoop loopBody actually runs. It's hard to test
-- the runLoop is isolation without restricting which parts run.

local function addRunLoopTaskToAllowlist( scheduler, runLoopTaskName )

	scheduler._testData.runLoopTaskAllowList[ runLoopTaskName ] = true

end

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

local function allowlistAllTasks( scheduler )

	scheduler._testData.runLoopTaskAllowList = nil

end

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

local function sendTestDataImp( scheduler, ... )

	if not scheduler._testData.testInfoBacklogPush then

		scheduler._testData.testInfoBacklogPush,
			scheduler._testData.testInfoBacklogPop = WFQueueing.makeQueue()

	end

	sendDataFromSchedulerThread( scheduler, scheduler._testData.testInfoStream,
									scheduler._testData.testInfoBacklogPush,
									scheduler._testData.testInfoBacklogPop,
	 								... )

end

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

local function expose_test_API( scheduler )

	if not scheduler._testData then 

		local testInfoStream = Rx.Subject()

		scheduler._testData = { 

			testInfoStream = testInfoStream,

			runLoopTaskAllowList = {},

		}

		rawset( scheduler, "_sendTestData", sendTestDataImp )

		rawset( scheduler, "_drainTestDataBacklog", sendTestDataImp )

	end

	return {

		suspendRunLoop = suspendRunLoop,

		resumeRunLoop = resumeRunLoop,

		addRunLoopTaskToAllowlist = addRunLoopTaskToAllowlist,

		allowlistAllTasks = allowlistAllTasks,

		runScheduledRequests = runScheduledRequests,

		setState = forwardToPriorityQueue( 'setState' ),

		hasRequestForLocalAssetId = 
			forwardToPriorityQueue( 'hasRequestForLocalAssetId' ),

		applyTransitionToAllWithState = applyTransitionToAllWithState,

		getRequestsWithStatesInPriorityOrder = getRequestsWithStatesInPriorityOrder,

	}

end

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

return {

	wrapInScheduleRequests = wrapInScheduleRequests,

	wrapSchedulerFunctionInBatchScheduleRequests = 
		wrapSchedulerFunctionInBatchScheduleRequests,

	wrapSchedulerFunctionInTransaction = wrapSchedulerFunctionInTransaction,

	getNextNRequestsWithState = getNextNRequestsWithState,

	getNextRequestWithState = getNextRequestWithState,

	-- for a given localAssetId, this only applies a transition to a request if
	-- the transition is valid for the request's current state.
	maybeApplyTransitionForLocalAssetId = 
		wrapSchedulerFunctionInBatchScheduleRequests( 
		"maybeApplyTransitionForLocalAssetId", 
		maybeApplyTransitionForLocalAssetId ),

	-- maybeSetNextState is more expensive that applyTransitionToRequest
	-- because it must lookup the request in the _requestPriorityQueue before
	-- updating the state.
	--
	-- That is needed in a worker thread callback because we have no guarantee
	-- the request still exists and if it does, no guarantee it had the same
	-- state as it did before we started the worker thread task
	maybeSetNextState = wrapInScheduleRequests( maybeSetNextState ),

	-- applyTransitionToRequest is much more efficient that maybeSetNextState
	-- and it should be used whenever you update can update a request without
	-- yielding to worker thread for further work.
	applyTransitionToRequest = applyTransitionToRequest,

	applyRequestTransitionOnScheduler = applyRequestTransitionOnScheduler,

	applyTransitionAllRequests = 
		wrapSchedulerFunctionInBatchScheduleRequests( 
			"applyTransitionAllRequests", applyTransitionAllRequests ),

	getRequestsWithStatesInPriorityOrder = 
		getRequestsWithStatesInPriorityOrder,

	mapAllRequestsWithPixStreamId = mapAllRequestsWithPixStreamId,

	mapAllRequestsWithStates = mapAllRequestsWithStates,

	mapAllRequestsWithState = mapAllRequestsWithState,

	reduceAllRequestsWithStates = reduceAllRequestsWithStates,

	sendRenditionPathToPixelStream = sendRenditionPathToPixelStream,

	sendLocalBinaryPathToPixelStream = sendLocalBinaryPathToPixelStream,

	sendErrorToPixelStream = sendErrorToPixelStream,

	countRequestsWithState = 
		forwardToPriorityQueue( 'countRequestsWithState' ),

	getLocalAssetIdRequestsWithState = 
		forwardToPriorityQueue( 'getLocalAssetIdRequestsWithState' ),

	getChangeCounter = forwardToPriorityQueue( 'getChangeCounter' ),

	getActiveLocalAssetIdSetCounter = 
		forwardToPriorityQueue( 'getActiveLocalAssetIdSetCounter' ),

	getStateChangeCounter = forwardToPriorityQueue( 'getStateChangeCounter' ),

	getLocalAssetIds = forwardToPriorityQueue( 'getLocalAssetIds' ),

	hasPixstreamRequestForLocalAssetId = 
		forwardToPriorityQueue( 'hasPixstreamRequestForLocalAssetId' ),

	wasRequestDeleted =
		forwardToPriorityQueue( 'wasRequestDeleted' ),

	getDeletedRequestsWithState =
		forwardToPriorityQueue( 'getDeletedRequestsWithState' ),

	getDeletedRequests =
		forwardToPriorityQueue( 'getDeletedRequests' ),

	purgeDeletedRequests =
		forwardToPriorityQueue( 'purgeDeletedRequests' ),

	getLocalAssetIdsWithNoCurrentRequests =
		forwardToPriorityQueue( 'getLocalAssetIdsWithNoCurrentRequests' ),

	purgeLocalAssetIdsWithNoCurrentRequests = 
		forwardToPriorityQueue( 'purgeLocalAssetIdsWithNoCurrentRequests' ),

	_DEBUG_getAllLocalAssetIdRequests =
		forwardToPriorityQueue( 'getAllLocalAssetIdRequests' ),

	expose_test_API = expose_test_API,

	sendTestData = function() end,

}
