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

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

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

local WFArguments = require 'WFArguments'
local WFArguments_check = assert( WFArguments.check, 
										"missing WFArguments.check" )
local WFCloseSupport = require "WFCloseSupport"
local WFFunctionUtils = require "WFFunctionUtils"
local WFLogger = require 'WFLogger'

	-- WF.sqlite.database
local WFSQLiteDatabase = require "WFSQLiteDatabase"
local WFSQLiteDatabaseSchema = require "WFSQLiteDatabaseSchema"
local WFSQLiteUtils = require 'WFSQLiteUtils'

	-- scheduler helpers
local PixelPathProviderRequestTypes = require "PixelPathProviderRequestTypes"

	-- Rx
local Rx = require "Rx"

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

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

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

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

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

local logger =  WFLogger 'PixelPathProviderPriorityQueue'

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

local KSDate_timeNowForAnimation = require "KSDate".timeNowForAnimation

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

local WFSQLiteUtils_createVariableArgList = assert( WFSQLiteUtils.createVariableArgList,
			"WFSQLiteUtils.createVariableArgList does not exist" )

--==============================================================================
-- utils
--==============================================================================

local function curry( f )

	local curried = {}

	return function( a )

		curried[ a ] = curried[ a ] or f( a )

		return curried[ a ]

	end

end

--==============================================================================
-- db queue
--==============================================================================

local function execSQL( db, sql, ... )

	db:execOnce( sql, ... )

	return db

end

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

local function makeDBQueue()

	local dummySchema = WFSQLiteDatabaseSchema( "1.0" )

	local db = WFSQLiteDatabase( ':memory:', dummySchema )

	local schemaSQL = {
		------------------------------------------------------------
		-- main Request table
		------------------------------------------------------------

		[[
		CREATE TABLE Request (
			pixStreamId NOT NULL,
			localAssetId NOT NULL, 
			requestPriority NOT NULL, 
			state NOT NULL,
			longSide NOT NULL,
			rowEpoch DEFAULT 0,
			lastStateUpdateEpoch DEFAULT 0
		);
		]],

		[[
		CREATE UNIQUE INDEX Request_primaryKey
						ON Request( pixStreamId, localAssetId );
		]],

		[[
		CREATE INDEX Request_byStateInPriorityOrder
						ON Request( state, requestPriority, rowEpoch, lastStateUpdateEpoch, localAssetId, longSide, pixStreamId );
		]],

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

		[[
		CREATE TABLE RequestLocalAssetIds (
			localAssetId PRIMARY KEY,
			requestCount NOT NULL
		);
		]],

		[[
		CREATE INDEX RequestLocalAssetIds_requestCount
						ON RequestLocalAssetIds( requestCount, localAssetId );
		]],

		[[
		CREATE TRIGGER AddToRequestLocalAssetIds
		AFTER INSERT ON RequestLocalAssetIds
		BEGIN

			INSERT INTO bump_activeLocalAssetIdSetCounter
				SELECT 1;

		END;
		]],

		[[
		CREATE TRIGGER RemoveFromRequestLocalAssetIds
		AFTER DELETE ON RequestLocalAssetIds
		BEGIN

			INSERT INTO bump_activeLocalAssetIdSetCounter
				SELECT 1;

		END;
		]],
		
		------------------------------------------------------------
		
		[[
		CREATE TABLE PriorityName (
			priority NOT NULL, 
			name NOT NULL
		);
		]],

		[[ 
		CREATE UNIQUE INDEX PriorityName_primaryKey
						ON PriorityName( priority, name );
		]],

		------------------------------------------------------------
		-- Track abandoned localAssetIds
		------------------------------------------------------------

		[[
		CREATE TABLE DeletedRequests(
			pixStreamId NOT NULL,
			localAssetId NOT NULL, 
			requestPriority NOT NULL, 
			state NOT NULL,
			longSide NOT NULL,
			rowEpoch DEFAULT 0,
			lastStateUpdateEpoch DEFAULT 0
		);
		]],

		[[
		CREATE UNIQUE INDEX DeletedRequests_primaryKey
						ON DeletedRequests( pixStreamId, localAssetId );
		]],

		[[ 
		CREATE INDEX DeletedRequests_lookupByState
						ON DeletedRequests( state, pixStreamId, localAssetId );
		]],

		[[
		CREATE TRIGGER Add_To_DeletedRequests
		AFTER DELETE ON Request
		BEGIN

			INSERT INTO bump_rowEpoch SELECT 1;

			UPDATE RequestLocalAssetIds
			SET requestCount = requestCount - 1
			WHERE localAssetId = OLD.localAssetId;

			INSERT OR REPLACE INTO DeletedRequests
				SELECT
					OLD.pixStreamId,
					OLD.localAssetId,
					OLD.requestPriority,
					OLD.state,
					OLD.longSide,
					OLD.rowEpoch,
					OLD.lastStateUpdateEpoch;

		END;
		]],

		[[
		CREATE TRIGGER update_Request_epochs_AFTER_INSERT
		AFTER INSERT ON Request 
		BEGIN

			UPDATE RequestLocalAssetIds
			SET requestCount = requestCount + 1
			WHERE localAssetId = NEW.localAssetId;

			INSERT OR REPLACE INTO RequestLocalAssetIds
				SELECT NEW.localAssetId, 1
					EXCEPT
				SELECT NEW.localAssetId, 1
				FROM RequestLocalAssetIds
				WHERE localAssetId = NEW.localAssetId;

			INSERT INTO bump_rowEpoch SELECT 1;

			INSERT INTO bump_lastStateUpdateEpoch SELECT 1;

			UPDATE Request
			SET rowEpoch = 
					( SELECT * FROM current_rowEpoch ),
			    lastStateUpdateEpoch = 
					( SELECT * FROM current_lastStateUpdateEpoch )
			WHERE pixStreamId = NEW.pixStreamId AND localAssetId = NEW.localAssetId;

		END;
		]],


		[[
		CREATE TRIGGER update_Request_state
		AFTER UPDATE OF state ON Request WHEN NEW.state != OLD.state
		BEGIN

			INSERT INTO bump_lastStateUpdateEpoch SELECT 1;

			UPDATE Request
			SET lastStateUpdateEpoch = 
					( SELECT * FROM current_lastStateUpdateEpoch )
			WHERE pixStreamId = NEW.pixStreamId AND localAssetId = NEW.localAssetId;

		END;
		]],


		------------------------------------------------------------
		-- change counter for ordering
		------------------------------------------------------------
		[[
		CREATE TABLE ChangeCounter(
			id PRIMARY KEY,
			changeCounter NOT NULL
		);
		]],

		[[
		CREATE VIEW current_rowEpoch AS
			SELECT changeCounter FROM ChangeCounter WHERE id = 'rowEpoch';
		]],

		[[
		CREATE VIEW current_lastStateUpdateEpoch AS
			SELECT changeCounter FROM ChangeCounter WHERE id = 'lastStateUpdateEpoch';
		]],

		[[
		CREATE VIEW bump_ChangeCounter AS
			SELECT 1 AS changeCounterId;
		]],

		[[
		CREATE TRIGGER bump_ChangeCounterImpl
		INSTEAD OF INSERT ON bump_ChangeCounter
		BEGIN

			INSERT OR REPLACE INTO ChangeCounter
				SELECT NEW.changeCounterId, IFNULL( MAX( changeCounter ), 0 ) + 1
				FROM ChangeCounter
				WHERE id = NEW.changeCounterId;

		END;
		]],

		[[
		CREATE VIEW bump_rowEpoch AS
			SELECT 1;
		]],

		[[
		CREATE TRIGGER bump_rowEpochImpl
		INSTEAD OF INSERT ON bump_rowEpoch
		BEGIN

			INSERT INTO bump_ChangeCounter SELECT 'rowEpoch';

		END;
		]],

		[[
		CREATE VIEW current_activeLocalAssetIdCounter AS
			SELECT changeCounter FROM ChangeCounter WHERE id = 'activeLocalAssetIdCounter';
		]],

		[[
		CREATE VIEW bump_activeLocalAssetIdSetCounter AS
			SELECT 1;
		]],

		[[
		CREATE TRIGGER bump_activeLocalAssetIdSetCounterImpl 
		INSTEAD OF INSERT ON bump_activeLocalAssetIdSetCounter
		BEGIN

			INSERT INTO bump_ChangeCounter SELECT 'activeLocalAssetIdCounter';

		END;
		]],

		[[
		CREATE VIEW bump_lastStateUpdateEpoch AS
			SELECT 1;
		]],

		[[
		CREATE TRIGGER bump_lastStateUpdateEpochImpl
		INSTEAD OF INSERT ON bump_lastStateUpdateEpoch
		BEGIN

			INSERT INTO bump_ChangeCounter SELECT 'lastStateUpdateEpoch';

		END;
		]],

	}

	db = WFFunctionUtils_reduce( db, execSQL, schemaSQL )

	for priority, name in pairs( PixelPathProviderRequestTypes.priorities ) do

		db:exec( [[INSERT INTO PriorityName SELECT ?, ?]], priority, name )

	end

	return db

end

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

local function maybeGetScalar( self, sql, ... )

	return self._db:gatherArray( sql, ... )[ 1 ]

end

--==============================================================================
-- PixelPathProviderPriorityQueue prototype
--==============================================================================

local PixelPathProviderPriorityQueuePrototype = {

	_db = false,
		-- an in memory db priority queue of all the pixel provider requests
	
	_defaultInitialState = false,
		-- the default initial state when creating a new request

	WFCloseSupport.addCloseSupportMethods,

}

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

function PixelPathProviderPriorityQueuePrototype:__init( defaultInitialState )

	WFArguments_check( "PixelPathProviderPriorityQueue", "string" )

	self._defaultInitialState = defaultInitialState

	self._db = makeDBQueue( defaultInitialState )

end

--==============================================================================
-- public object methods
--==============================================================================

local noop = function() 
	logger:warnf( "\n\nMethod called after the object has closed.n\n" )
	return nil
end

local function noopDb()

	return {
		exec = noop,
		gatherArray = noop,
		gatherArrayOfArray = noop,
	}

end

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

function PixelPathProviderPriorityQueuePrototype:close()

	self._db:close()
	self._db = noopDb()

end

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

function PixelPathProviderPriorityQueuePrototype:maybeGetCurrState(
													pixStreamId, localAssetId )

	return maybeGetScalar( self,
							[[SELECT state
							  FROM Request
							  WHERE pixStreamId = ? AND localAssetId = ?
							  LIMIT 1]],
							  pixStreamId, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:setState( pixStreamId, localAssetId, 
															newState )

	self._db:exec( [[UPDATE Request
							SET state = ?
							WHERE pixStreamId = ? AND localAssetId = ?]],
						newState,
						pixStreamId, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:setNextStateForLocalAssetIdWithState(
									sourceState, localAssetId, nextState )

	self._db:exec( [[UPDATE Request
							SET state = ?
							WHERE state = ? AND localAssetId = ?]],
						nextState, sourceState, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:changeStateForAllWithState( 
													oldState, newState )

	self._db:exec( [[UPDATE Request
							SET state = ?
							WHERE state = ?]],
						newState,
						oldState )

end

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

local function getRequestsWithStatesInPriorityOrderSQL( nStates )

	return string.format(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
			  FROM Request
			  WHERE state IN ( %s )
			  ORDER BY state, requestPriority, rowEpoch]], 
			  WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

getRequestsWithStatesInPriorityOrderSQL = curry( getRequestsWithStatesInPriorityOrderSQL )

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

function PixelPathProviderPriorityQueuePrototype:getRequestsWithStatesInPriorityOrder( ... )

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

	assert(  nStates > 0, "one or more states required" )

	return self._db:gatherArrayOfArray( 
					getRequestsWithStatesInPriorityOrderSQL( nStates ), 
					... )

end

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

local PixelPathProviderRequestTypes_getPriority = 
		PixelPathProviderRequestTypes.getPriority

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

local function getValidatedPriority( requestPriorityName )

	local requestPriority = PixelPathProviderRequestTypes_getPriority(
														requestPriorityName )

	if not requestPriority then

		error( string.format( "%s is not a valid request requestPriority",
									requestPriorityName ) )

	end

	return requestPriority

end

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

local function _DEBUG_recordStartTimerForPixStream( self, pixStreamId )

	local startTimes = 
			rawget( self, "_DEBUG_pixStreamActiveRequestsStartTime" ) or {} 

	startTimes[ pixStreamId ] = startTimes[ pixStreamId ] or 
									KSDate_timeNowForAnimation()

	rawset( self, "_DEBUG_pixStreamActiveRequestsStartTime", startTimes )

end

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

local PixelPathProviderStateMachine_EndStates

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

local function _DEBUG_isEndState( state )

	PixelPathProviderStateMachine_EndStates =
		PixelPathProviderStateMachine_EndStates or
		require "PixelPathProviderStateMachine".EndStates

	return PixelPathProviderStateMachine_EndStates[ state ] ~= nil

end

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

function PixelPathProviderPriorityQueuePrototype:addNewRequest( pixStreamId,
													localAssetId, requestPriorityName,
													longSide, 
													maybeOverrideInitialState )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "number", "string", "number|string", "?string" )

	local requestPriority = getValidatedPriority( requestPriorityName )

	local initialState = maybeOverrideInitialState or self._defaultInitialState

	if not _DEBUG_isEndState( initialState ) then

		_DEBUG_recordStartTimerForPixStream( self, pixStreamId )

	end

	self._db:exec(
		[[INSERT OR REPLACE INTO Request( 
						pixStreamId,
						localAssetId,
						requestPriority,
						state,
						longSide )
		  VALUES(
						?,
						?,
						?,
						?,
						? )]],
		pixStreamId,
		localAssetId, 
		requestPriority, 
		initialState,
		longSide )

end

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

local function maybeGetLongSide( self, pixStreamId, localAssetId )

	return maybeGetScalar( self,
							[[SELECT longSide 
							  FROM Request
							  WHERE pixStreamId = ? AND localAssetId = ?
							  LIMIT 1]],
							  pixStreamId, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:maybeUpdateRequest( 
													pixStreamId, localAssetId,
													requestPriorityName, longSide )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "number", "string", "number|string" )

	local requestPriority = getValidatedPriority( requestPriorityName )

	local prevLongSide = maybeGetLongSide( self, pixStreamId, localAssetId )

	if not prevLongSide then

		return false

	end

	self._db:exec(
		[[UPDATE Request
		  SET requestPriority = ?, longSide = ?
		  WHERE pixStreamId = ? AND localAssetId = ? AND 
		  		( requestPriority != ? OR longSide != ? )]],
		requestPriority, longSide,
		pixStreamId, localAssetId,
		requestPriority, longSide )

	return true, prevLongSide

end

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

function PixelPathProviderPriorityQueuePrototype:maybeUpdateLongSide( 
													pixStreamId, localAssetId,
													longSide )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "number", "number|string" )

	local prevLongSide = maybeGetLongSide( self, pixStreamId, localAssetId )

	if not prevLongSide then

		return false

	end

	if prevLongSide == longSide then

		return false

	end

	self._db:exec(
		[[UPDATE Request
		  SET longSide = ?
		  WHERE pixStreamId = ? AND localAssetId = ?]],
		longSide,
		pixStreamId, localAssetId )

	return true, prevLongSide

end

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

local function getAllPixstreamIdRequests( self, pixStreamId )

	return self._db:gatherArrayOfArray(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
			  FROM Request
			  WHERE pixStreamId = ?]],
			pixStreamId ) 

end

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

PixelPathProviderPriorityQueuePrototype.getAllPixstreamIdRequests =
	getAllPixstreamIdRequests

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

local function getNextNRequestsWithStateSQL( nStates )

	return string.format(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
			  FROM Request
			  WHERE state IN ( %s )
			  ORDER BY state, requestPriority, rowEpoch
			  LIMIT ?]], WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

getNextNRequestsWithStateSQL = curry( getNextNRequestsWithStateSQL )

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

function PixelPathProviderPriorityQueuePrototype:getNextNRequestsWithState( 
														n, ... )

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

	assert(  nStates > 0, "one or more states required" )

	local statesPlusLimit = { ... }

	statesPlusLimit[ #statesPlusLimit + 1 ] = n

	return self._db:gatherArrayOfArray(
					getNextNRequestsWithStateSQL( nStates ),
					unpack( statesPlusLimit ) )

end

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

local function getNextNRequestsWithPriorityAndStatesSQL( nStates )

	return string.format(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
			  FROM Request
			  WHERE state IN ( %s ) AND requestPriority = ?
			  ORDER BY state, requestPriority, rowEpoch
			  LIMIT ?]], WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

getNextNRequestsWithPriorityAndStatesSQL = 
	curry( getNextNRequestsWithPriorityAndStatesSQL )

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

function PixelPathProviderPriorityQueuePrototype:getNextNRequestsWithPriorityAndStates( 
														n, priority, ... )

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

	assert(  nStates > 0, "one or more states required" )

	local statesPlusPriorityAndLimit = { ... }

	statesPlusPriorityAndLimit[ #statesPlusPriorityAndLimit + 1 ] = priority

	statesPlusPriorityAndLimit[ #statesPlusPriorityAndLimit + 1 ] = n

	return self._db:gatherArrayOfArray(
					getNextNRequestsWithPriorityAndStatesSQL( nStates ),
					unpack( statesPlusPriorityAndLimit ) )

end

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

local function countRequestsWithStateSQL( nStates )

	return string.format( [[SELECT COUNT( * )
							  FROM Request
							  WHERE state IN ( %s )]], 
							  	WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

countRequestsWithStateSQL = curry( countRequestsWithStateSQL )

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

function PixelPathProviderPriorityQueuePrototype:countRequestsWithState( ... )

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

	assert(  nStates > 0, "one or more states required" )

	return maybeGetScalar( self, countRequestsWithStateSQL( nStates ), ... )

end

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

local function getLocalAssetIdRequestsWithStateSQL( nStates )

	return string.format(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
				FROM Request
				WHERE localAssetId = ? AND state IN ( %s )
				ORDER BY state, requestPriority, rowEpoch]], 
				WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

getLocalAssetIdRequestsWithStateSQL = 
	curry( getLocalAssetIdRequestsWithStateSQL )

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

function PixelPathProviderPriorityQueuePrototype:getLocalAssetIdRequestsWithState( 
													localAssetId, ... )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"number" )

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

	assert(  nStates > 0, "one or more states required" )

	return self._db:gatherArrayOfArray( 
					getLocalAssetIdRequestsWithStateSQL( nStates ),
					localAssetId, ... )

end

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

function PixelPathProviderPriorityQueuePrototype:removeRequest( pixStreamId,
																localAssetId )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "number" )

	self._db:exec( [[DELETE FROM Request 
						WHERE pixStreamId = ? AND localAssetId = ?]], 
					pixStreamId, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:batchRemoveRequests( pixStreamId,
																localAssetIds )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "table" )

	assert( #localAssetIds <= 64 )

	local sql = string.format(
		[[DELETE FROM Request WHERE pixStreamId = ? AND localAssetId IN ( %s )]],
		WFSQLiteUtils_createVariableArgList( #localAssetIds ) )

	self._db:exec( sql, pixStreamId, unpack( localAssetIds ) )

end

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

function PixelPathProviderPriorityQueuePrototype:setRequestPriorityToNewPriority( 
													pixStreamId, localAssetId,
													currRequestTypeName,
													newRequestTypeName )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "number", "string", "string" )

	local currRequestPriority = getValidatedPriority( currRequestTypeName )

	local newRequestPriority = getValidatedPriority( newRequestTypeName )

	self._db:exec( [[UPDATE Request 
						SET requestPriority = ?
						WHERE pixStreamId = ? AND localAssetId = ? AND
								requestPriority = ?]], 
					newRequestPriority, pixStreamId, localAssetId,
					currRequestPriority )

end

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

function PixelPathProviderPriorityQueuePrototype:updateLongSideOnAll( 
													pixStreamId, 
													newLongSide )

	WFArguments_check( "PixelPathProviderPriorityQueue",
						"string", "string|number" )

	self._db:exec( [[UPDATE Request 
						SET longSide = ?
						WHERE pixStreamId = ?]], 
					newLongSide, pixStreamId )

	return getAllPixstreamIdRequests( self, pixStreamId )

end

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

function PixelPathProviderPriorityQueuePrototype:getAllLocalAssetIdRequests( 
													localAssetId )

	WFArguments_check( "PixelPathProviderPriorityQueue", "number" )

	return self._db:gatherArrayOfArray(
			[[SELECT pixStreamId, localAssetId, longSide, requestPriority, state
			  FROM Request
			  WHERE localAssetId = ?]],
			localAssetId ) 

end

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

function PixelPathProviderPriorityQueuePrototype:setPriority( pixStreamId,
																localAssetId,
																newPriority )

	WFArguments_check( "PixelPathProviderPriorityQueue",
							"string", "number", "string" )

	newPriority = getValidatedPriority( newPriority )

	self._db:exec( [[UPDATE Request 
						SET requestPriority = ?
						WHERE pixStreamId = ? AND localAssetId = ?]], 
					newPriority, pixStreamId, localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:abandonAllPixStreamRequests( 
												pixStreamId )

	self._db:exec( [[DELETE FROM Request WHERE pixStreamId = ?]], pixStreamId )

end

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

function PixelPathProviderPriorityQueuePrototype:abandonAllLocalAssetIdRequests( 
												localAssetId )

	self._db:exec( [[DELETE FROM Request WHERE localAssetId = ?]], localAssetId )

end

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

function PixelPathProviderPriorityQueuePrototype:getChangeCounter()

	return maybeGetScalar( self, [[SELECT * FROM current_rowEpoch]] )

end

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

function PixelPathProviderPriorityQueuePrototype:getActiveLocalAssetIdSetCounter()

	return maybeGetScalar( 
			self, [[SELECT * FROM current_activeLocalAssetIdCounter]] )

end

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

function PixelPathProviderPriorityQueuePrototype:getStateChangeCounter()

	return maybeGetScalar( self, [[SELECT * FROM current_lastStateUpdateEpoch ]] )

end

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

function PixelPathProviderPriorityQueuePrototype:getLocalAssetIdsForPixstreamId( 
														pixStreamId )

	return self._db:gatherArray( 
					[[SELECT localAssetId 
					  FROM Request 
					  WHERE pixStreamId = ?]], 
					pixStreamId )

end

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

function PixelPathProviderPriorityQueuePrototype:hasPixstreamRequestForLocalAssetId( 
														pixStreamId, 
														localAssetId )

	return maybeGetScalar( self,
							[[SELECT localAssetId
							  FROM Request 
							  WHERE pixStreamId = ? AND localAssetId = ?
							  LIMIT 1]], 
							pixStreamId, localAssetId ) ~= nil

end

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

function PixelPathProviderPriorityQueuePrototype:hasRequestForLocalAssetId( 
														localAssetId )

	return maybeGetScalar( self,
							[[SELECT localAssetId
							  FROM Request 
							  WHERE localAssetId = ?
							  LIMIT 1]], 
							localAssetId ) ~= nil

end

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

function PixelPathProviderPriorityQueuePrototype:getLocalAssetIds()

	return self._db:gatherArray( 
			[[SELECT localAssetId FROM RequestLocalAssetIds
			  WHERE requestCount > 0]] )

end

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

function PixelPathProviderPriorityQueuePrototype:getLocalAssetIdsWithNoCurrentRequests()

	return self._db:gatherArray( 
			[[SELECT localAssetId FROM RequestLocalAssetIds
			  WHERE requestCount <= 0]] )

end

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

function PixelPathProviderPriorityQueuePrototype:purgeLocalAssetIdsWithNoCurrentRequests()

	return self._db:gatherArray( 
			[[DELETE FROM RequestLocalAssetIds
			  WHERE requestCount <= 0]] )

end

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

function PixelPathProviderPriorityQueuePrototype:wasRequestDeleted( 
													pixstreamId, localAssetId )

	return maybeGetScalar( self,
				[[SELECT COUNT( * ) 
				  FROM DeletedRequests
				  WHERE pixStreamId = ? AND localAssetId = ?]],
				pixstreamId, localAssetId ) > 0

end

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

function PixelPathProviderPriorityQueuePrototype:getDeletedRequests()

	return self._db:gatherArrayOfArray( 
				[[SELECT pixStreamId, localAssetId FROM DeletedRequests]] )

end

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

local function getDeletedRequestsWithStateSQL( nStates )

	return string.format( [[SELECT pixStreamId, localAssetId, state 
							FROM DeletedRequests
							WHERE state IN ( %s )]],
								WFSQLiteUtils_createVariableArgList( nStates ) )

end

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

getDeletedRequestsWithStateSQL = curry( getDeletedRequestsWithStateSQL )

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

function PixelPathProviderPriorityQueuePrototype:getDeletedRequestsWithState( ... )

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

	return self._db:gatherArrayOfArray( 
						getDeletedRequestsWithStateSQL( nStates ), ... )

end

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

function PixelPathProviderPriorityQueuePrototype:purgeDeletedRequests()

	return self._db:exec( [[DELETE FROM DeletedRequests]] )

end

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

local _priorityNameMap

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

local function addPriorityNameMapEntry( priorityNameMap, priorityData )

	local name, requestPriority = unpack( priorityData )

	priorityNameMap[ requestPriority ] = name

	return priorityNameMap

end

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

local function getPriorityNameMap( self )

	if not _priorityNameMap then

		_priorityNameMap = 
			WFFunctionUtils_reduce(
				{}, addPriorityNameMapEntry,
						self._db:gatherArrayOfArray( [[
							SELECT name, priority 
							FROM PriorityName]] ) )

	end

	return _priorityNameMap
	
end

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

local function formatPixStreamIdStateRow( pixStreamIdStateRow, priorityNameMap )

	local requestPriority, state, count = unpack( pixStreamIdStateRow )

	local priorityName = priorityNameMap[ requestPriority ] or
							"unknown_priority"

	return { string.format( "%s_%s", state, priorityName ), count }

end

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

local function sortCountRowByName( countRow1, countRow2 )

	return countRow1[ 1 ] < countRow2[ 1 ]

end

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

local function processPixStreamIdStateRow( results, pixStreamIdStateRow, 
											priorityNameMap )

	table.insert( results.stateCounts, 
					formatPixStreamIdStateRow( pixStreamIdStateRow,
													priorityNameMap ) )

	local _, state = unpack( pixStreamIdStateRow )

	results.hasNonEndState =
		results.hasNonEndState or not _DEBUG_isEndState( state )

	return results

end

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

local function _DEBUG_updateDuration( self, pixStreamId )

	local startTimes = 
			rawget( self, "_DEBUG_pixStreamActiveRequestsStartTime" ) or {} 

	if startTimes[ pixStreamId ] then

		local start = startTimes[ pixStreamId ]

		startTimes[ pixStreamId ] = nil

		rawset( self, "_DEBUG_pixStreamActiveRequestsStartTime", startTimes )

		local durations = 
				rawget( self, "_DEBUG_pixStreamActiveRequestsDuration" ) or {} 

		durations[ pixStreamId ] = ( durations[ pixStreamId ] or 0 ) +
										KSDate_timeNowForAnimation() - start

		rawset( self, "_DEBUG_pixStreamActiveRequestsDuration", durations )

	end

end

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

local function addPixStreamIdStateCounts( countsPerPixStreamId, pixStreamId, 
											self, priorityNameMap )

	local rawStateCounts =
			self._db:gatherArrayOfArray( [[
					SELECT requestPriority, state, COUNT( * ) 
					FROM Request
					WHERE pixStreamId = ?
					GROUP BY requestPriority, state]], pixStreamId ) 

	local results = WFFunctionUtils_reduce( 
							{ hasNonEndState = false, stateCounts = {} },
							processPixStreamIdStateRow,
								rawStateCounts, priorityNameMap )

	if not results.hasNonEndState then

		-- no active requests left
		_DEBUG_updateDuration( self, pixStreamId )

	end

	local stateCounts = results.stateCounts

	table.sort( stateCounts, sortCountRowByName )

	local totalPixStreamIdRequests = 
				maybeGetScalar( 
							self, 
							[[SELECT COUNT( * ) 
							  FROM Request
							  WHERE pixstreamId = ?]], pixStreamId )

	stateCounts[ #stateCounts + 1 ] = { "total", totalPixStreamIdRequests }

	countsPerPixStreamId[ pixStreamId ] = stateCounts

	return countsPerPixStreamId

end

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

local function maybeAddRequestPriorityIndex( self )

	-- this SQLite INDEX is only used to speed up debugging APIs

	if not rawget( self, "_addedRequestPriorityIndex" ) then

		rawset( self, "_addedRequestPriorityIndex", true )

		self._db:exec(

			[[
			CREATE INDEX IF NOT EXISTS Request_byPriorityOrder
							ON Request( requestPriority, state, localAssetId, lastStateUpdateEpoch, rowEpoch, longSide, pixStreamId );
			]] )

	end

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getRequestCountsByState()

	maybeAddRequestPriorityIndex( self )

	local priorityNameMap = getPriorityNameMap( self )

	local pixStreamIds = 
		self._db:gatherArray( 
			[[SELECT DISTINCT pixStreamId FROM Request]] )

	local countsPerPixStreamId = 
			WFFunctionUtils_reduce( {},
				addPixStreamIdStateCounts,
					pixStreamIds, self, priorityNameMap )

	for pixStreamId in pairs( rawget( self, "_DEBUG_pixStreamActiveRequestsStartTime" ) or {} ) do

		if not countsPerPixStreamId[ pixStreamId ] then
			
			-- no requests left, record the duration
			_DEBUG_updateDuration( self, pixStreamId )
			
		end
	end

	local rawStateCountsAllRequests =
			self._db:gatherArrayOfArray( [[
					SELECT requestPriority, state, COUNT( * ) 
					FROM Request
					GROUP BY requestPriority, state]] )

	local stateCountsAllRequests = WFFunctionUtils_map( 
							formatPixStreamIdStateRow,
								rawStateCountsAllRequests, priorityNameMap )

	table.sort( stateCountsAllRequests, sortCountRowByName )

	local totalRequests = maybeGetScalar( 
							self, [[SELECT COUNT( * ) FROM Request]] )

	stateCountsAllRequests[ #stateCountsAllRequests + 1 ] = 
		{ "total", totalRequests }

	countsPerPixStreamId.all = stateCountsAllRequests

	return countsPerPixStreamId

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getRequestsByPriority( 
													requestPriorityName )

	-- This is a full table scan, it will be slow, use sparingly (only for 
	-- diagnostic log)

	local requestPriority = getValidatedPriority( requestPriorityName )

	return self._db:gatherArrayOfArray( [[
				SELECT pixStreamId, localAssetId, longSide, requestPriority, state
				FROM Request
				WHERE requestPriority = ? 
				ORDER BY pixstreamId, localAssetId]],
				requestPriority )

end

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

local function formatPriorityCountRow( priorityCountRow, priorityNameMap )

	local requestPriority, count = unpack( priorityCountRow )

	return { priorityNameMap[ requestPriority ] or "unknown_priority", count }

end

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

local function addPriorityCountsForPixStreamId( 
				pixStreamIdPriorityCounts, pixStreamId, self, priorityNameMap )

	local rawPriorityCounts =
			self._db:gatherArrayOfArray( [[
					SELECT requestPriority, COUNT( * ) 
					FROM Request
					WHERE pixstreamId = ?
					GROUP BY requestPriority
					ORDER BY requestPriority]], pixStreamId ) 

	local priorityCounts = WFFunctionUtils_map( 
							formatPriorityCountRow,
								rawPriorityCounts, priorityNameMap )

	local totalRequests = maybeGetScalar( 
							self, 
							[[SELECT COUNT( * ) 
							  FROM Request
							  WHERE pixstreamId = ?]], pixStreamId )

	priorityCounts[ #priorityCounts + 1 ] = { "total", totalRequests }

	pixStreamIdPriorityCounts[ pixStreamId ] = priorityCounts 

	return pixStreamIdPriorityCounts
end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getPixStreamActiveRequestDurations()

	return rawget( self, "_DEBUG_pixStreamActiveRequestsDuration" ) or {}

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getRequestCountsByPriority()

	maybeAddRequestPriorityIndex( self )

	local priorityNameMap = getPriorityNameMap( self )

	local pixStreamIds = 
		self._db:gatherArray( 
			[[SELECT DISTINCT pixStreamId FROM Request]] )

	local countsPerPixStreamId = 
			WFFunctionUtils_reduce( {},
				addPriorityCountsForPixStreamId,
					pixStreamIds, self, priorityNameMap )

	local rawPriorityCountsAllRequests =
			self._db:gatherArrayOfArray( [[
					SELECT requestPriority, COUNT( * ) 
					FROM Request
					GROUP BY requestPriority
					ORDER BY requestPriority]] )

	local priorityCountsAllRequest = WFFunctionUtils_map( 
							formatPriorityCountRow,
								rawPriorityCountsAllRequests, priorityNameMap )

	local totalRequests = maybeGetScalar( 
							self, [[SELECT COUNT( * ) FROM Request]] )

	priorityCountsAllRequest[ #priorityCountsAllRequest + 1 ] = 
		{ "total", totalRequests }

	countsPerPixStreamId.all = priorityCountsAllRequest

	return countsPerPixStreamId

end

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

local function initStateBenchmarkRow( db, stateName )

	return execSQL( 
			db, 
			[[INSERT OR REPLACE INTO RequestStateBenchmark SELECT ?, 0, 0]],
			stateName )

end

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

local function addStateBenchmarking( db, stateNames )

	local stateBenchmarkingSQL = {
		------------------------------------------------------------
		-- main Request table
		------------------------------------------------------------

		[[
		CREATE TABLE RequestStateTimer (
			pixStreamId NOT NULL,
			localAssetId NOT NULL, 
			stateLastSet NOT NULL
		); 
		]],

		[[
		CREATE UNIQUE INDEX RequestStateTimer_primaryKey
						ON RequestStateTimer( pixStreamId, localAssetId );
		]],

		[[
		CREATE TABLE RequestStateBenchmark(
			state PRIMARY KEY,
			elapsed DEFAULT 0,
			occurrences DEFAULT 0
		);
		]],

		[[
		CREATE TRIGGER update_RequestStateTimer_AFTER_Request_INSERT
		AFTER INSERT ON Request 
		BEGIN

			INSERT OR REPLACE INTO RequestStateTimer
				SELECT NEW.pixStreamId, NEW.localAssetId, julianday( 'now' ) * 86400;

			INSERT INTO UpdateRequestStateTransitions 
				SELECT 'NewRequest', NEW.state;

			DELETE FROM DeletedRequests
			WHERE pixStreamId = NEW.pixStreamId and localAssetId = NEW.localAssetId;

		END;
		]],


		[[
		CREATE TRIGGER update_RequestStateTimer_AFTER_Request_state_UPDATE
		AFTER UPDATE OF state ON Request WHEN NEW.state != OLD.state
		BEGIN

			UPDATE RequestStateBenchmark
			SET 
				elapsed = elapsed +
							IFNULL( (
								SELECT julianday( 'now' ) * 86400 - stateLastSet
								FROM RequestStateTimer
								WHERE pixStreamId = NEW.pixStreamId AND
										localAssetId = NEW.localAssetId ),
								0 ),
				occurrences = occurrences + 1
			WHERE state = OLD.state; 

			INSERT OR REPLACE INTO RequestStateTimer 
			SELECT NEW.pixStreamId, NEW.localAssetId, julianday( 'now' ) * 86400;

			INSERT INTO UpdateRequestStateTransitions 
				SELECT OLD.state, NEW.state;

		END;
		]],

		[[
		CREATE TABLE RequestStateTransitions(
			startState NOT NULL,
			nextState NOT NULL,
			occurrences NOT NULL
		);
		]],

		[[
		CREATE UNIQUE INDEX RequestStateTransitions_primaryKey
						ON RequestStateTransitions( nextState, startState );
		]],

		[[
		CREATE VIEW UpdateRequestStateTransitions AS
			SELECT 1 AS startState, 1 AS nextState;
		]],

		[[
		CREATE TRIGGER UpdateRequestStateTransitionsImpl 
		INSTEAD OF INSERT ON UpdateRequestStateTransitions
		BEGIN

			INSERT OR REPLACE INTO RequestStateTransitions 
				SELECT NEW.startState, NEW.nextState, 
						IFNULL( 
							( SELECT occurrences + 1
							  FROM RequestStateTransitions
							  WHERE startState = NEW.startState AND 
							  		nextState = NEW.nextState ),
							1 );

		END;
		]],


	}

	db = WFFunctionUtils_reduce( db, execSQL, stateBenchmarkingSQL )

	-- add records for all state machine states
	db = WFFunctionUtils_reduce( db, initStateBenchmarkRow, stateNames )

	-- add records for all requests
	db = execSQL( db,
			[[INSERT OR REPLACE INTO RequestStateTimer
				  SELECT pixStreamId, localAssetId, julianday( 'now' ) * 86400
				  FROM Request ]] )

	return db

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_enableRequestStateBenchmarking( stateNames )

	if not rawget( self, "_DEBUG_requestStateBenchmarkingEnabled" ) then

		rawset( self, "_DEBUG_requestStateBenchmarkingEnabled", true )

		self._db = addStateBenchmarking( self._db, stateNames )

	end

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getRequestStateBenchmarks()

	if not rawget( self, "_DEBUG_requestStateBenchmarkingEnabled" ) then

		return

	end

	return self._db:gatherArrayOfArray( 
			[[SELECT state, elapsed, occurrences
			  FROM RequestStateBenchmark
			  WHERE elapsed > 0 OR occurrences > 0]] )

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getRequestStateTransitionCounts(
													stateName )

	if not rawget( self, "_DEBUG_requestStateBenchmarkingEnabled" ) then

		return

	end

	return self._db:gatherArrayOfArray( 
			[[SELECT startState, nextState, occurrences
			  FROM RequestStateTransitions
			  WHERE nextState = ? AND occurrences > 0]], stateName )

end

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

function PixelPathProviderPriorityQueuePrototype:_DEBUG_getPixStreamIdStatePairs( 
													localAssetId,
													stateChangeCounter )


	return self._db:gatherArrayOfArray( [[
					SELECT pixStreamId, state
					FROM Request
					WHERE localAssetId IN (
						SELECT localAssetId
						FROM Request
						WHERE localAssetId = ? AND 
								lastStateUpdateEpoch > ? )]],
					localAssetId, stateChangeCounter )

end

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

local function pack( ... )

	return { ... }, select( "#", ... )

end

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

function PixelPathProviderPriorityQueuePrototype:runInTransaction( f, ... )

	self._db:exec( [[SAVEPOINT runInTransaction]] )

	local ret, retN = pack( pcall( f, ... ) )

	if not ret[ 1 ] then
	
		self._db:exec( [[ROLLBACK]] )

		error( ret[ 2 ] )

	end

	self._db:exec( [[RELEASE runInTransaction]] )

	return unpack( ret, 2, retN )

end

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

local PixelPathProviderPriorityQueue = 
		Rx.Class( "PixelPathProviderPriorityQueue", 
					PixelPathProviderPriorityQueuePrototype )

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

return PixelPathProviderPriorityQueue 
