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

WFRxDocStoreChangesFeedImp.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2017 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 Rx = require "Rx"
local Observable = Rx.Observable

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

	-- WF.core
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFRequestDispatcher = require 'WFRequestDispatcher'

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

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

	-- WF.sqlite.doc_store
local WFRxDocStoreUtils = require "WFRxDocStoreUtils"

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

local check = assert( WFArguments.check, "check does not exist" )

local WFRxDocStoreUtils_processChangedDocsRawDocRows = WFRxDocStoreUtils.processChangedDocsRawDocRows
local WFRxDocStoreUtils_unpackDocValue = WFRxDocStoreUtils.unpackDocValue

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

local logger = require 'WFLogger'( 'WFRxDocStoreChangesFeedImp' )

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

local Rx_Observable_create = Observable.create
local Rx_Observable_combineLatest = Observable.combineLatest
local Rx_Observable_concat = Observable.concat
local Rx_Observable_defer = Observable.defer
local Rx_Observable_empty = Observable.empty
local Rx_Observable_from = Observable.from
local Rx_Observable_just = Observable.just
local Rx_Observable_merge = Observable.merge

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

local Rx_BehaviorSubject = Rx.BehaviorSubject
local Rx_Subject = Rx.Subject

local kMaxChangeCounter = math.huge

--==============================================================================
-- @mark docValue change counter support
--==============================================================================

local function docValueChangeCounter( docValue )

	return tonumber( docValue._changeCounter )

end

--==============================================================================
-- @mark changes buffer
--==============================================================================

local function createChangesFeedBuffer( maxSize )

	check( 'number' )

	return { _maxSize = maxSize, _wrapped = false, _tailOffset = 0, 
				_headOffset = 0, _bufferArray = {}, _mostRecentRevIdByDocId = {} }

end

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

local function getBufferDocValue( buffer, bufferOffset )

	return buffer._bufferArray[ bufferOffset ]

end

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

local function getBufferOffset( buffer, deltaFromHeadOffset )

	local maxOffset = buffer._maxSize

	local delta = deltaFromHeadOffset % maxOffset

	local headOffset = buffer._headOffset

	return headOffset + delta <= maxOffset and 
				headOffset + delta or
				headOffset + delta - maxOffset

end

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

local function incrementBufferOffset( buffer, offset )

	return offset < buffer._maxSize and offset + 1 or 1

end

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

local function formRevIdAndChangeCounterStr( revId, changeCounter )

	-- We track which document revision is the most recent by looking at both
	-- the document revision Id, as well as the change counter associated with the
	-- change. We use both fields because in the event of a conflict, the
	-- revision Ids might match.

	return string.format( "%s:%s", revId, changeCounter )

end

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

local function isMostRecentDocumentRevision( mostRecentRevIdByDocId, docValue )
	
	if not docValue or not docValue._id then
		return false
	end
	
	local mostRecentRevId = mostRecentRevIdByDocId[ docValue._id ]
	
	if not mostRecentRevId then
		return true
	end
	
	return mostRecentRevId == formRevIdAndChangeCounterStr( docValue._revId, docValue._changeCounter )

end

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

local function maybeStopTrackingMostRecentDocumentRevision( mostRecentRevIdByDocId, docValue )

	if isMostRecentDocumentRevision( mostRecentRevIdByDocId, docValue ) then
		mostRecentRevIdByDocId[ docValue._id ] = nil
	end
	
	return mostRecentRevIdByDocId

end

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

local function trackMostRecentDocumentRevision( mostRecentRevIdByDocId, docValue )
	
	if docValue and docValue._id then
		mostRecentRevIdByDocId[ docValue._id ] = formRevIdAndChangeCounterStr( docValue._revId, docValue._changeCounter )
	end
	
	return mostRecentRevIdByDocId
	
end

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

local function appendDocValueToBuffer( buffer, docValue )

	buffer._wrapped = buffer._wrapped or buffer._tailOffset == buffer._maxSize

	buffer._tailOffset = incrementBufferOffset( buffer, buffer._tailOffset )

	local previousDocValueAtOffset = buffer._bufferArray[ buffer._tailOffset ]
		
	-- if we have wrapped the ring buffer, check to see if the "popped" docValue
	-- is the most recent known docValue in _bufferArray, if so, remove it from the 
	-- _mostRecentRevIdByDocId as well so that _mostRecentRevIdByDocId doesn't grow 
	-- unbounded. (i.e. _mostRecentRevIdByDocId should always be a subset of _bufferArray)

	buffer._mostRecentRevIdByDocId =
			maybeStopTrackingMostRecentDocumentRevision( 
					buffer._mostRecentRevIdByDocId, previousDocValueAtOffset )
	
	buffer._mostRecentRevIdByDocId = 
			trackMostRecentDocumentRevision( 
					buffer._mostRecentRevIdByDocId, docValue )

	buffer._bufferArray[ buffer._tailOffset ] = docValue

	buffer._headOffset = not buffer._wrapped and 
							1 or 
							incrementBufferOffset( buffer, buffer._tailOffset )
	
	return buffer

end

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

local function bufferSize( buffer )

	if buffer._headOffset == 0 then

		return 0

	end

	if not buffer._wrapped then

		return buffer._tailOffset

	end

	return buffer._maxSize

end

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

local function getChangedDocsFromBufferOffset( changesFeedBuffer, bufferOffset )

	local buf = changesFeedBuffer._bufferArray

	local currOffset = bufferOffset

	local tailOffset = changesFeedBuffer._tailOffset

	local ret = {}

	repeat

		ret[ #ret + 1 ] = buf[ currOffset ]

		if currOffset == tailOffset then

			return ret

		end

		currOffset = incrementBufferOffset( changesFeedBuffer, currOffset )

	until false

end

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

local function getChangeCounterAtBufferOffset( changesFeedBuffer,
												bufferOffset )

	return docValueChangeCounter( 
			getBufferDocValue( changesFeedBuffer, bufferOffset ) )

end

--------------------------------------------------------------------------------
-- getBufferOffsetForChangeCounter returns one of the following
-- 
-- 	nil, nil
-- 		means the buffer is empty
-- 
-- 	offset
-- 		is the earliest offset in the non-empty changesBuffer such that all
-- 		earlier offsets have change counters less than sinceChangeCounter
-- 
-- 	nil, -1 
-- 		means the buffer is non-empty and sinceChangeCounter is before the
-- 		change counter of the head element
-- 		
-- 	nil, 1
-- 		means the buffer is non-empty and sinceChangeCounter is after change
-- 		counter of the tail element
--------------------------------------------------------------------------------

local function getBufferOffsetForChangeCounter( changesFeedBuffer,
														sinceChangeCounter )

	local nBuffer = bufferSize( changesFeedBuffer )

	if nBuffer == 0 then

		return nil, nil

	elseif nBuffer == 1 then
	
		-- there is only one element in the buffer.

		local changeCounter = 
				getChangeCounterAtBufferOffset(
					changesFeedBuffer, changesFeedBuffer._headOffset )
					
		if sinceChangeCounter > changeCounter then
			
			-- sinceChangeCounter is later than the element in the buffer.
			
			return nil, 1

		elseif sinceChangeCounter == changeCounter then
		
			return changesFeedBuffer._headOffset
		
		end

		-- sinceChangeCounter is earlier than the element in the buffer.

		return nil, -1
		
	end

	-- do a binary search for the change counter
				
	local lowDeltaFromHeadOffset = 0

	local highDeltaFromHeadOffset = nBuffer - 1

	while lowDeltaFromHeadOffset < highDeltaFromHeadOffset do

		local midpointDeltaFromHeadOffset = 
				math.floor( 
					( lowDeltaFromHeadOffset + highDeltaFromHeadOffset )/ 2 )

		local midpointBufferOffset = getBufferOffset(
										changesFeedBuffer, 
										midpointDeltaFromHeadOffset )

		local midpointChangeCounter = 
				getChangeCounterAtBufferOffset( 
					changesFeedBuffer, midpointBufferOffset )

		if midpointChangeCounter > sinceChangeCounter then

			highDeltaFromHeadOffset = midpointDeltaFromHeadOffset - 1

		elseif midpointChangeCounter < sinceChangeCounter then

			lowDeltaFromHeadOffset = midpointDeltaFromHeadOffset + 1

		else

			return midpointBufferOffset

		end

	end

	if lowDeltaFromHeadOffset == nBuffer - 1 then

		-- all elements except the tail have been checked and are earlier than
		-- sinceChangeCounter. check the tail to determine final result.

		local tailChangeCounter = 
				getChangeCounterAtBufferOffset(
					changesFeedBuffer, changesFeedBuffer._tailOffset )

		if sinceChangeCounter <= tailChangeCounter then
			
			return changesFeedBuffer._tailOffset

		end

		return nil, sinceChangeCounter < tailChangeCounter and -1 or 1

	elseif highDeltaFromHeadOffset == -1 then

		-- this happens when sinceChangeCounter was earlier than buffer head
		return nil, -1

	elseif highDeltaFromHeadOffset == 0 then

		-- all elements except the head have been checked and are later than
		-- sinceChangeCounter. check the head to determine final result.

		local headChangeCounter = 
				getChangeCounterAtBufferOffset(
					changesFeedBuffer, changesFeedBuffer._headOffset )

		if sinceChangeCounter >= headChangeCounter then
			
			return changesFeedBuffer._headOffset

		end

		return nil, -1

	end

	local currBufferOffset = getBufferOffset( changesFeedBuffer,
												highDeltaFromHeadOffset )

	local currChangeCounter =
			getChangeCounterAtBufferOffset( 
				changesFeedBuffer, currBufferOffset )
	
	return currChangeCounter >= sinceChangeCounter and
			currBufferOffset or 
			incrementBufferOffset( changesFeedBuffer, currBufferOffset )
	
end

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

local function getChangeCounterBeforeHead( changesFeedBuffer )

	return getChangeCounterAtBufferOffset( 
			changesFeedBuffer, changesFeedBuffer._headOffset ) - 1
					
end

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

local function getChangeCounterAfterTail( changesFeedBuffer )

	return getChangeCounterAtBufferOffset( 
				changesFeedBuffer, changesFeedBuffer._tailOffset ) + 1

end

--------------------------------------------------------------------------------
--
-- maybeGetBufferedChanges returns
--	
--		nil, nil, nil 
--		
--			when the changesFeedBuffer is empty
--		
--		nil, changedDocs, futureChangeCounter 
--		
--			when all sinceChangeCounter changedDocs are in the buffer.
--			
--				changedDocs is the changedDocs from sinceChangeCounter to now
--				
--				futureChangeCounter is the change counter from which new
--				relevant changes will start
--			
--		beforeChangeCounter, changedDocs, futureChangeCounter
--		
--			when some sinceChangeCounter changedDocs are in the buffer, but
--			not all. 
--			
--				beforeChangeCounter is the upper bound of past changedDocs
--				that must still be fetched
--			
--				changedDocs is the changedDocs from beforeChangeCounter + 1 to
--				now
--				
--				futureChangeCounter is the change counter from which new
--				relevant changes will start
--			
--			counter from which new relevant changes will start
--		
--		nil, nil, futureChangeCounter 
--		
--			when sinceChangeCounter is after all buffered changed docs.
--			
--				futureChangeCounter is the change counter from which new
--				relevant changes will start

local function maybeGetBufferedChanges( changesFeedBuffer, sinceChangeCounter )

	local maybeOffset, maybeChangeCounterDeltaFromSince =
			getBufferOffsetForChangeCounter( changesFeedBuffer, 
												sinceChangeCounter )

	if maybeOffset == nil and maybeChangeCounterDeltaFromSince == nil then

		return nil, nil, nil

	end

	local futureChangeCounter = getChangeCounterAfterTail( changesFeedBuffer )

	if maybeOffset then

		-- all the changes are buffered
		return nil,
				getChangedDocsFromBufferOffset( changesFeedBuffer, 
													maybeOffset ),
				futureChangeCounter

	end

	if maybeChangeCounterDeltaFromSince < 0 then

		-- some changes buffered, but still need catchup
		return getChangeCounterBeforeHead( changesFeedBuffer ),
				getChangedDocsFromBufferOffset( 
					changesFeedBuffer, changesFeedBuffer._headOffset ),
				futureChangeCounter

	end

	-- requested changes from the future
	return nil, nil, math.max( futureChangeCounter, sinceChangeCounter )

end

--==============================================================================
-- @mark docValueChangeArraySinceUntilOnce
--==============================================================================

local _processChangedDocsRawDocRows = {

	fn = WFRxDocStoreUtils_processChangedDocsRawDocRows
		-- The above is parameterized solely to support unit test overrides.

}

--------------------------------------------------------------------------------
-- makeContinueFromStreamAndPump returns a changeCounterStream and a function
-- for pushing new values onto the stream
--------------------------------------------------------------------------------

local function makeContinueFromStreamAndPump( initialContinueFrom )

	local continueFromStream = Rx_BehaviorSubject( initialContinueFrom )

	local function sendNewContinueFromValue( continueFrom )

		continueFromStream:onNext( continueFrom )

	end 


	return continueFromStream, sendNewContinueFromValue

end 

--------------------------------------------------------------------------------
-- makeContinueFromChangedDocsStream returns a stream of changedDocs arrays 
-- for each element that is emitted on the continueFromStream
--------------------------------------------------------------------------------

local function makeContinueFromChangedDocsStream( 
					docStoreImp, continueFromStream, untilChangeCounter )

	return continueFromStream
			:flatMapLatest(
				function( continueFrom )

					if continueFrom > untilChangeCounter then
						
						return Rx_Observable_just( {} )

					end

					local continueFromArg =
							{ fromChangeCounter = continueFrom }

					return docStoreImp
							:request( 'changedDocs', continueFromArg, 'changesFeedImp' )
							:map( function( rawDocRows, changeCounter, limit )

								return _processChangedDocsRawDocRows.fn(
										rawDocRows, changeCounter, limit, true, {} )

							end )							
							:take( 1 )
				end )
			:share()

end

-------------------------------------------------------------------------------
-- makeSignalSawAllDocsUntilChangeCounter creates a semaphore like object and a
-- function for signaling it.
-------------------------------------------------------------------------------

local function makeSignalSawAllDocsUntilChangeCounter()

	local sawAllDocsUntilChangeCounter = Rx_Subject()

	local function maybeSignalSawAllDocsUntilChangeCounter( 
						changedDocs, nowChangeCounter )

		if #changedDocs == 0 or 
				changedDocs[ 1 ]._changeCounter > nowChangeCounter then

			sawAllDocsUntilChangeCounter:onNext( true )

		end

	end

	return sawAllDocsUntilChangeCounter, maybeSignalSawAllDocsUntilChangeCounter

end

--------------------------------------------------------------------------------
-- given an array of changed docs, emits a stream of changed docs that
-- occurred before nowChangeCounter. If there additional changed docs that
-- occur after this array, this will call sendNewContinueFromValue after 
-- emitting the last item in the changedDocsArray
--------------------------------------------------------------------------------

local function changedDocsNotAfterUntilStream( changedDocs, untilChangeCounter, _,
											sendNewContinueFromValue )

	-- want to pull changed docs from a batch until we see docs after now
	local i = 0

	local function getNextChangedDocUntilChangeCounter()

		i = i + 1

		if i > #changedDocs then

			return nil

		end

		local changedDoc = changedDocs[ i ] 

		if changedDoc._changeCounter > untilChangeCounter then
			
			return nil

		end

		return changedDoc

	end

	return Rx_Observable_from( getNextChangedDocUntilChangeCounter )
			:tapOnCompleted(
				function()

					sendNewContinueFromValue( changedDocs.continueFrom )

				end )

end

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

local function changedDocsSinceUntil( docStoreImp, sinceChangeCounter, untilChangeCounter )

	local continueFromStream, sendNewContinueFromValue =
			makeContinueFromStreamAndPump( sinceChangeCounter )

	local sawAllDocsUntilChangeCounter, 
			maybeSignalSawAllDocsUntilChangeCounter =
				makeSignalSawAllDocsUntilChangeCounter()

	local continueFromChangedDocsStream = 
			makeContinueFromChangedDocsStream( 
				docStoreImp, continueFromStream, untilChangeCounter )

	local ongoingChangedDocsStream =
			Rx_Observable_combineLatest( 
					continueFromChangedDocsStream,
					Rx_Observable_just( untilChangeCounter ),
					Rx_Observable_just( sawAllDocsUntilChangeCounter ),
					Rx_Observable_just( sendNewContinueFromValue ) )

	return ongoingChangedDocsStream
			:tap( maybeSignalSawAllDocsUntilChangeCounter )
			:takeUntil( sawAllDocsUntilChangeCounter )
			:flatMapLatest( changedDocsNotAfterUntilStream )
			:share()

end

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

local function changeCounterNotLessThan( changeCounter )

	return function( docValue )

		return docValue._changeCounter >= changeCounter

	end

end

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

local function futureChangesFeed( self, futureChangeCounter )

	return self._sharedChangesFeedSinceNow
			:filter( changeCounterNotLessThan( futureChangeCounter ) )

end

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

local function makeCatchupChangedDocsOnceStreams( self, sinceChangeCounter,
													untilChangeCounter,
													futureChangeCounter )

	if sinceChangeCounter > untilChangeCounter then

		return Rx_Observable_empty(), Rx_Observable_empty()

	end

	futureChangeCounter = futureChangeCounter or untilChangeCounter + 1

	local catchupChangedDocsCompleted = Rx.Subject()

	local catchupChangedDocs =
			changedDocsSinceUntil( self._docStoreImp,
				sinceChangeCounter,
				untilChangeCounter )
			:tapOnCompleted( function()
				catchupChangedDocsCompleted:onNext( true )
			end )

	local accumulatedWhileCatchingUpChangedDocsOnce =
			futureChangesFeed( self, futureChangeCounter )
				:takeUntil( catchupChangedDocsCompleted )

	return catchupChangedDocs, accumulatedWhileCatchingUpChangedDocsOnce

end

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

local function assembleCatchupChangedDocsStream(
				self, sinceChangeCounter, origFutureChangeCounter,
				buffereredChangedDocsArray, catchUpChangedDocs,
				accumulatedWhileCatchingUpChangedDocs )

	-- Assemble a stream by concatenating the changed docs before the buffer
	-- (the catchUpChangedDocs), the buffered docs, the docs accumulated from
	-- the changes feed while producing the catchUpChangedDocs and the future
	-- changes feed.

	local futureChangeCounter = origFutureChangeCounter or sinceChangeCounter

	local lastDocValue = {}
	local accumulatedWhileCatchingUpBuffer = {}
	return catchUpChangedDocs
			:merge( accumulatedWhileCatchingUpChangedDocs
					:toArray()
					:tap( function( array )

						-- save off the accumulated values so we can append
						-- them onto the stream after the docs from the buffer,
						-- and then make sure this observable doesn't emit
						-- any values until then via takeLast( 0 ).
						accumulatedWhileCatchingUpBuffer = array

						local catchupCount = #array

						local totalCatchUpCount = self._debugChangesCounts[ 'totalCatchUpCount' ] or 0
						self._debugChangesCounts[ 'totalCatchUpCount' ] = totalCatchUpCount + catchupCount

						local maxCatchUpCount = self._debugChangesCounts[ 'maxCatchUpCount' ] or 0
						self._debugChangesCounts[ 'maxCatchUpCount' ] = math.max( maxCatchUpCount, catchupCount )

						if #array > 100 then
							logger:warnf( "accumulatedWhileCatchingUpChangedDocs created a large array of docValues size = %d", #array )
						end

					end )
					:takeLast(0))
		:concat( Rx_Observable_from( buffereredChangedDocsArray ) )
		:concat( Rx_Observable_defer( function()

			-- Defer creation of observable until accumulatedWhileCatchingUpBuffer
			-- is populated.
			return Rx_Observable_from( accumulatedWhileCatchingUpBuffer )
				:tapOnCompleted( function()
					accumulatedWhileCatchingUpBuffer = nil
				end )

		end ) )
		:tap( function( docValue )
			lastDocValue = docValue
		end )
		:concat( Rx_Observable_defer( function()

			-- Defer creation of futureChangesFeed until lastDocValue contains
			-- the last doc value seen.
			futureChangeCounter = math.max( futureChangeCounter,
				lastDocValue._changeCounter or 0 )

			return futureChangesFeed( self, futureChangeCounter )

		end ) )

end

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

local function catchupChangedDocsStream(
			self, sinceChangeCounter, bufferedSinceToNowChangedDocs,
			beforeChangeCounterOnce, futureChangeCounter )


	return beforeChangeCounterOnce
		:flatMapLatest( function( beforeChangeCounter )

			local catchUpChangedDocs, accumulatedWhileCatchingUpChangedDocs =
				makeCatchupChangedDocsOnceStreams( self,
					sinceChangeCounter, beforeChangeCounter, futureChangeCounter )

			return assembleCatchupChangedDocsStream( self,
				sinceChangeCounter,
				futureChangeCounter,
				bufferedSinceToNowChangedDocs,
				catchUpChangedDocs,
				accumulatedWhileCatchingUpChangedDocs )

		end )

end

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

local function nowChangeCounterOnce( self )

	return self._docStoreImp:request( 'changeCounter' ):take( 1 )

end

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

local function changesFeedSinceStream( self, sinceChangeCounter )

	local maybeBeforeChangeCounter, maybeBufferedSinceToNowChangedDocs,
			maybeFutureChangeCounter = 
				maybeGetBufferedChanges( 
					self._changesFeedBuffer, sinceChangeCounter )

	local beforeChangeCounterOnce =
			-- first check if no changes have been seen on self yet. If that's
			-- the case we might need to do a catch up query
			( maybeFutureChangeCounter == nil and 
				nowChangeCounterOnce( self ) ) or
			-- if we have a beforeChangeCounter we definitely need a catch up
			( maybeBeforeChangeCounter and
					Rx_Observable_just( maybeBeforeChangeCounter ) )

	local bufferedSinceToNowChangedDocs = 
			maybeBufferedSinceToNowChangedDocs or {}

	if beforeChangeCounterOnce then

		-- we might need to do a catch up query
		return catchupChangedDocsStream(
				self, sinceChangeCounter, bufferedSinceToNowChangedDocs,
				beforeChangeCounterOnce,
				maybeFutureChangeCounter )

	end

	-- we don't need a catchup query just stream out the buffered changes (if
	-- any) followed by all future changes
	return Rx_Observable_concat(
				Rx_Observable_from( bufferedSinceToNowChangedDocs ),
				futureChangesFeed( self, maybeFutureChangeCounter ) )

end

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

local WFRxDocStoreChangesFeedImp = {

	_docStoreImp = false,
		-- (WFWorkerThread) reference to the worker thread
	
	_sharedChangesFeedSinceNow = false,
		-- Rx.Disposable subscription to changes feed
	
	_changesFeedBuffer = false,
		-- lua table with object semantics that holds n most recent docValues 
		-- from a _raw_changesFeed
		
	_changesFeedBufferUpdaterSubscription = false,

	_docValueWatchedIds = false,
		-- (table) indexed by docValueWatcherIds that contains maps for
		-- watchedDocIds with key=docId, value=changeCounter seen.

	_docValueWatcherSubject = false,
		-- (table) indexed by docValueWatcherIds that contain subjects to
		-- redirect values from changes to the watchedIds.

	_debugChangesCounts = false,
		-- table of counts processed indexed by ns_funcName hash.

	WFCloseSupport.addCloseSupportMethods,

}

--------------------------------------------------------------------------------
		
local function makeDocStoreImpWorker( path )

	return WFWorkerThread( "WFRxDocStoreMessageProcessor.lua", path) 

end

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

local _docStoreImpCreator = {

	f = makeDocStoreImpWorker
		-- The above is parameterized solely to support unit test overrides.

}

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

local _changesFeedBufferConfig = {

	size = 50,
		-- The above is parameterized solely to support unit test overrides.

}

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

function WFRxDocStoreChangesFeedImp:__init( path )

	self:setOwned { 

		_docStoreImp = _docStoreImpCreator.f( path )

	}

	self._sharedChangesFeedSinceNow = self._docStoreImp
											:request( '_raw_changesFeed' )
											:flatMap( function( rawDocRows, changeCounter, limit )

												local docValues = _processChangedDocsRawDocRows.fn(
														rawDocRows, changeCounter, limit, true, {} )

												return Rx.Observable.from( docValues )

											end )
											:share()

	self._changesFeedBuffer = createChangesFeedBuffer( 
								_changesFeedBufferConfig.size  )

	self._changesFeedBufferUpdaterSubscription = 
			self._sharedChangesFeedSinceNow
				:scan( appendDocValueToBuffer, self._changesFeedBuffer )
				:subscribe( 
					function( changesFeedBuffer )
						self._changesFeedBuffer = changesFeedBuffer

					end )

	self._docValueWatchedIds = {}
	self._docValueWatcherSubject = {}

	self._debugChangesCounts = {}

	WFDiagnosticLog.addLogProvider( self )

end

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

function WFRxDocStoreChangesFeedImp:sendDiagnosticLogData()

	check( 'WFRxDocStoreChangesFeedImp' )

	return {

		name = "WFRxDocStoreChangesFeedImp",
		id = 'WFRxDocStoreChangesFeedImp',
		weight = 1610,
		parentSection = WFDiagnosticLog.Sections.DOC_STORE,

		body = self._debugChangesCounts,

	}

end

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

function WFRxDocStoreChangesFeedImp:close()

	self._changesFeedBufferUpdaterSubscription:dispose()

end

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

local function maybeLoadFunc( ns, funcName )

	local ok, maybeNS = pcall( require, ns )

	if not ok or type( maybeNS ) ~= "table" then

		local errStr = 
				string.format( "Failed to load namespace %s: %s",
									ns, 
									not ok and maybeNS or 
										"namespace did not return a table" )

		logger:warn( errStr )

		return errStr

	end

	local maybeFunc = maybeNS[ funcName ]

	if( type( maybeFunc ) ~= "function" ) then

		local errStr = string.format( "No function at %s.%s", ns, funcName )

		logger:warn( errStr )

		return errStr

	end

	return maybeFunc

end

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

local _cachedDynamicallyLoadedFuncs = {}

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

local function getCachedFunc( ns, funcName )

	check( 'string', 'string' )

	_cachedDynamicallyLoadedFuncs[ ns ] = 
		_cachedDynamicallyLoadedFuncs[ ns ] or {}

	_cachedDynamicallyLoadedFuncs[ ns ][ funcName ] =
		_cachedDynamicallyLoadedFuncs[ ns ][ funcName ] or 
		maybeLoadFunc( ns, funcName )

	return _cachedDynamicallyLoadedFuncs[ ns ][ funcName ]

end

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

local function pack( ... )

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

end

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

local function scannedFilteredMappedChangesFeedSince(
							self, sinceChangeCounter, 
							namespace, scanFilterMapFuncName,
							initialScanState, emitStaleDocumentRevisions, optDesc )

	local scanFilterMapFunc = getCachedFunc( namespace, scanFilterMapFuncName )

	if type( scanFilterMapFunc) ~= "function" then

		error( string.format( "Failed to load %s.%s", 
								namespace, scanFilterMapFuncName ) )

	end

	local scanState = initialScanState

	local nameHash = string.format( '%s %s', namespace, optDesc or scanFilterMapFuncName )
	local nameHashSent = nameHash .. '_sent'

	local debugChangesCounts = self._debugChangesCounts

	debugChangesCounts[ nameHash ] = debugChangesCounts[ nameHash ] or 0
	debugChangesCounts[ nameHashSent ] = debugChangesCounts[ nameHashSent ] or 0

	return Rx_Observable_create(

		function( observer )

			return changesFeedSinceStream( self, sinceChangeCounter )
					:subscribe(

						function( docValue )
						
							if not emitStaleDocumentRevisions then
							
								local changesFeedBuffer = self._changesFeedBuffer or {}
								local mostRecentRevIdByDocId = changesFeedBuffer._mostRecentRevIdByDocId or {}
								
								if not isMostRecentDocumentRevision( mostRecentRevIdByDocId, docValue ) then
									
									return
								
								end
								
							end

							debugChangesCounts[ nameHash ] = debugChangesCounts[ nameHash ] + 1

							local n, ret = pack( pcall( 
											scanFilterMapFunc, 
												scanState, docValue ) )

							if not ret[ 1 ] then

								observer:onError( ret[ 2 ] )

								return

							end

							scanState = ret[ 2 ]

							if n > 2 then

								debugChangesCounts[ nameHashSent ] = debugChangesCounts[ nameHashSent ] + 1

								observer:onNext( unpack( ret, 3, n ) )

							end

						end,

						function( errStr )

							observer:onError( errStr )

						end,

						function()

							observer:onCompleted()

						end )

			end )

end

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

function WFRxDocStoreChangesFeedImp:scannedFilteredMappedChangesFeedSince(
										sinceChangeCounter, 
										namespace, scanFilterMapFuncName,
										initialScanState, optDesc )
										
	check( 'WFRxDocStoreChangesFeedImp', 'number', 'string', 'string', '_', '?string' )

	local emitStaleDocumentRevisions = true

	return scannedFilteredMappedChangesFeedSince(
						self, sinceChangeCounter, 
						namespace, scanFilterMapFuncName,
						initialScanState, emitStaleDocumentRevisions, optDesc )
end

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

function WFRxDocStoreChangesFeedImp:scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions(
										sinceChangeCounter, 
										namespace, scanFilterMapFuncName,
										initialScanState )
										
	check( 'WFRxDocStoreChangesFeedImp', 'number', 'string', 'string' )

	local emitStaleDocumentRevisions = false

	return scannedFilteredMappedChangesFeedSince(
						self, sinceChangeCounter, 
						namespace, scanFilterMapFuncName,
						initialScanState, emitStaleDocumentRevisions )
						
end

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

function WFRxDocStoreChangesFeedImp:makeDocValueWatcher( docValueWatcherId, optDesc )

	-- This stream is used exclusively to control the lifetime of the
	-- docValueWatcher.

	check( 'WFRxDocStoreChangesFeedImp', 'number', '?string' )

	-- Initialize diagnosticLog change counters if needed.
	local nameHash = optDesc and "DocWatcher" .. "_" .. optDesc or "DocWatcher"
	local nameHashSent = nameHash .. '_sent'
	local nameHashMake = nameHash .. '_TimesMade'

	local debugChangesCounts = self._debugChangesCounts
	debugChangesCounts[ nameHash ] = debugChangesCounts[ nameHash ] or 0
	debugChangesCounts[ nameHashSent ] = debugChangesCounts[ nameHashSent ] or 0
	debugChangesCounts[ nameHashMake ] = ( debugChangesCounts[ nameHashMake ] or 0 ) + 1

	-- Create a map to track the docValue changeCounter last sent.
	self._docValueWatchedIds[ docValueWatcherId ] = {}

	local docValueWatcherSubject = Rx_Subject()
	self._docValueWatcherSubject[ docValueWatcherId ] = docValueWatcherSubject

	return Rx_Observable_merge( docValueWatcherSubject, self._sharedChangesFeedSinceNow )

	:filter( function( docValue )

		local debugChangesCounts = self._debugChangesCounts
		debugChangesCounts[ nameHash ] = debugChangesCounts[ nameHash ] + 1

		-- Only return items in watchedDocIdSet and if it's _changeCounter
		-- is large than the set's current value.

		local watchedDocIdMap = self._docValueWatchedIds[ docValueWatcherId ]

		-- If watchedDocIdMap is empty this will default to '_id'
		local docIdName = type( next( watchedDocIdMap ) ) == 'number' and '_localId' or '_id'

		-- Get the requested _localId or _id as specified in docName, add
		-- a special case for _empty docValues as they may put the _localId in the _id field.
		local id = docValue[ docIdName ] or docValue._id

		-- Send any doc where the map entry is less than the current change counter
		-- along with empty docs as they don't have a _changeCounter.
		local sendDoc = not docValue._changeCounter
				or ( watchedDocIdMap[ id ] or kMaxChangeCounter ) < docValue._changeCounter

		if sendDoc then

			-- update the changeCounter for this docId
			watchedDocIdMap[ id ] = docValue._changeCounter

			debugChangesCounts[ nameHashSent ] = debugChangesCounts[ nameHashSent ] + 1

		end

		return sendDoc

	end )

	-- Clear out the _docValueWatchedIds entry for this
	-- when the subscription is disposed.
	:finally( function()
		self._docValueWatchedIds[ docValueWatcherId ] = nil
		self._docValueWatcherSubject[ docValueWatcherId ] = nil
	end )

end

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

local function getMissingDocsInMap( watchedIds, docIds )

	local missingIds = {}

	for _, id in ipairs( docIds ) do
		if not watchedIds[ id ] then
			missingIds[ #missingIds + 1 ] = id
		end
	end

	return missingIds

end

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

local function copyWatchedIdsInArray( watchedIds, docIds )

	-- Create a new map of watchedIds that only contains docIds that are in the
	-- docIds array.

	local newSetOfIds = {}

	for _, id in ipairs( docIds ) do
		newSetOfIds[ id ] = watchedIds[ id ]
	end

	return newSetOfIds

end

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

local function addDocsToWatchedIds( watchedIds, docIds, changeCounter )

	for _, id in ipairs( docIds ) do
		watchedIds[ id ] = changeCounter
	end

end

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

function WFRxDocStoreChangesFeedImp:updateDocValueWatcher( docValueWatcherId, docIdsToWatch, optOnlyChanges, optDesc )

	check( 'WFRxDocStoreChangesFeedImp', 'number', 'table', '?boolean', '?string' )

	local currentlyWatchedIds = self._docValueWatchedIds[ docValueWatcherId ]

	-- If this assert fails return Rx_Observable_empty() instead
	assert( currentlyWatchedIds )

	-- Copy the docIds we are currently watching (removes old entries).
	currentlyWatchedIds = copyWatchedIdsInArray( currentlyWatchedIds, docIdsToWatch )

	-- Get the new ids in docIdsToWatch that are not currently watched.
	local addedDocIds = getMissingDocsInMap( currentlyWatchedIds, docIdsToWatch )

	-- Add the new ids to the currently watched ids.
	addDocsToWatchedIds( currentlyWatchedIds, addedDocIds, 0 )

	-- Set the new set of currentlyWatchedIds into the array.
	self._docValueWatchedIds[ docValueWatcherId ] = currentlyWatchedIds

	if #addedDocIds == 0 or optOnlyChanges then
		return Rx_Observable_empty()
	end

	local docValueWatcherSubject = self._docValueWatcherSubject[ docValueWatcherId ]

	-- Get the doc value for the new ids being watched.
	return self._docStoreImp:request( 'bulkDocValue', addedDocIds, nil, 'updateDocValueWatcher' )
		:flatMap( Rx_Observable_from )
		:map( WFRxDocStoreUtils_unpackDocValue )
		:filter( function( docValue )

			-- Forward the values to the docWatcher stream, filtering out
			-- everything from this stream, but allow it to complete to
			-- finish the requested message.
			if docValueWatcherSubject and not docValueWatcherSubject:isDisposed() then
				docValueWatcherSubject:onNext( docValue )
			end

			return false

		end )

end

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

local WFRxScheduler

function WFRxDocStoreChangesFeedImp:_DEBUG_changesFeedInfoStream( refreshSeconds )

	check( 'WFRxDocStoreChangesFeedImp', 'number' )

	WFRxScheduler = WFRxScheduler or require "WFRxScheduler"

	return Rx.Observable.timer( 0, refreshSeconds, WFRxScheduler.sharedScheduler() )
		:map( function()
			return self._debugChangesCounts
		end )

end

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

WFRxDocStoreChangesFeedImp.DEBUG_dumpCoverage = function( _, path )

	check( 'WFRxDocStoreChangesFeedImp', '+string' )

	require 'SetupCoverage'.dumpCoverage( path )

	return Rx_Observable_empty()

end

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

WFRxDocStoreChangesFeedImp.class = {}

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

function WFRxDocStoreChangesFeedImp.class.initMessageProcessor( Class, path )

	local wfRxDocStoreChangesFeedImp = Class( path )

	WFCloseSupport.mustCloseAtThreadTermination( wfRxDocStoreChangesFeedImp )

	WFRequestDispatcher.onMessageDo {
		message = "scannedFilteredMappedChangesFeedSince",
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.onMessageDo {
		message = "scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions",
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.onMessageDo {
		message = "makeDocValueWatcher",
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.onMessageDo {
		message = "updateDocValueWatcher",
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.onMessageDo {
		message = '_DEBUG_changesFeedInfoStream',
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_dumpCoverage',
		target = wfRxDocStoreChangesFeedImp,
	}

	WFRequestDispatcher.enableDiagnosticLogForRequestStats()

end

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

function WFRxDocStoreChangesFeedImp.class._expose_test_API()

	return {

		createChangesFeedBuffer = createChangesFeedBuffer,

		appendDocValueToBuffer = appendDocValueToBuffer,

		getBufferOffset = getBufferOffset,

		getBufferOffsetForChangeCounter = getBufferOffsetForChangeCounter,

		maybeGetBufferedChanges = maybeGetBufferedChanges,

		getChangedDocsFromBufferOffset = getChangedDocsFromBufferOffset,

		changesFeedSinceStream = changesFeedSinceStream,

		injectAlternateDocStoreImpCreator = 

			function( alternateDocStoreImpCreator )

				local prevCreator = _docStoreImpCreator.f

				_docStoreImpCreator.f = alternateDocStoreImpCreator

				return prevCreator

			end,

		injectAlternateProcessChangedDocsRawDocRow = 

			function( alternateProcessChangedDocsRawDocRowsFn )

				local prevRawDocFn = _processChangedDocsRawDocRows.fn

				_processChangedDocsRawDocRows.fn = alternateProcessChangedDocsRawDocRowsFn

				return prevRawDocFn

			end,

		injectAlternateBufferSize = 
			function( newBufferSize )

				local prevSize = _changesFeedBufferConfig.size

				_changesFeedBufferConfig.size = newBufferSize 

				return prevSize

			end,

	}

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

return Rx.Class 'WFRxDocStoreChangesFeedImp'( WFRxDocStoreChangesFeedImp )
