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

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

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

WFRxDocValueWatcher consumes a sequence of array-valued tables of document IDs
and reports the values of each corresponding document either as it first appears
in the array, or as it changes in value while still in the array.

Document values may be either numeric (local) or string (global) IDs.

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

	-- RxLua
local Rx = require 'Rx'

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

	-- WF.core.utils
local KSTableUtils = require "KSTableUtils"
local WFFunctionUtils = require "WFFunctionUtils"
local WFArguments = require 'WFArguments'
local WFRxBatchedBulkOperation = require 'WFRxBatchedBulkOperation'

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

local WFFunctionUtils_reduce = WFFunctionUtils.reduce

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

local KSTableUtils_deepCopy = KSTableUtils.deepCopy

local KSTableUtils_appendArray = KSTableUtils.appendArray

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

local function reportDocValueIfWatched( currWatchedDocIds, docValue )
		
	if currWatchedDocIds[ docValue._id ] or currWatchedDocIds[ docValue._localId ] then

		return currWatchedDocIds, docValue

	end

	return currWatchedDocIds

end

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

local function addToSet( set, element )

	set[ element ] = true

	return set

end

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

local function removeFromSet( set, element )

	set[ element ] = nil

	return set

end

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

local function make( docIdArrayStream, docStore )

	WFArguments.check( '_', 'WFRxDocStore' )

	local currWatchedDocIds = {}
	local docIdsBeingFetchedArray = {}

	local continueFromStream = Rx.BehaviorSubject()

	local function updateContinueFromStream( docValue )
		
		if not tonumber( docValue._changeCounter ) then
			-- _empty docValues may not have a _changeCounter
			-- since these are deleted or don't exit and 
			-- we shouldn't need to track these further
			return
		end
		
		local currContinueFrom = continueFromStream:getValue()

		local changeCounterAfter = docValue._changeCounter + 1

		local newContinueFrom =
				currContinueFrom and
					math.max( currContinueFrom, changeCounterAfter ) or
					changeCounterAfter

		if newContinueFrom ~= currContinueFrom then

			continueFromStream:onNext( newContinueFrom )

		end

	end

	local function getChangesFeedForDocIds( docIdSet )

		local myDocIdSet = KSTableUtils_deepCopy( docIdSet )

		return Rx.Observable.create(
			function( observer )

				return continueFromStream
					:take( 1 )
					:flatMapLatest(
						function( maybeContinueFrom )

							local changesFromContinueFrom =
									maybeContinueFrom and
										docStore
											:scannedFilteredMappedChangesFeedSince( 
												"WFRxDocValueWatcher", "reportDocValueIfWatched", 
												myDocIdSet, maybeContinueFrom ) or
										docStore
											:scannedFilteredMappedChangesFeed( 
												"WFRxDocValueWatcher", "reportDocValueIfWatched", 
												myDocIdSet )

							return changesFromContinueFrom
									:tap( updateContinueFromStream )

						end )
					:subscribe( observer )

			end )

	end

	return WFRxArrayDiffer( docIdArrayStream )
			:flatMapLatest( 
				function( newDocIdsArray, removedDocIdsArray )

					if not next( currWatchedDocIds ) and
							#docIdsBeingFetchedArray == 0 and
							#newDocIdsArray == 0 and 
							#removedDocIdsArray == 0 then

						return Rx.Observable.empty()

					end

					currWatchedDocIds = 
						WFFunctionUtils_reduce( 
							currWatchedDocIds, 
							removeFromSet,
								removedDocIdsArray )


					local unfetchedDocIdsSet = {}

					unfetchedDocIdsSet =
						WFFunctionUtils_reduce(
							unfetchedDocIdsSet, addToSet, docIdsBeingFetchedArray )

					unfetchedDocIdsSet =
						WFFunctionUtils_reduce(
							unfetchedDocIdsSet, removeFromSet, removedDocIdsArray )


					if #newDocIdsArray == 0 and next( unfetchedDocIdsSet ) == nil then

						return getChangesFeedForDocIds( 
								currWatchedDocIds )

					end

					-- Preserve order of fetched docIds.
					-- Add ids from prior array diff that didn't complete.
					local newDocIdsToFetchArray = {}
					for _, docId in ipairs( docIdsBeingFetchedArray ) do
						if unfetchedDocIdsSet[ docId ] then
							newDocIdsToFetchArray[ #newDocIdsToFetchArray + 1 ] = docId
						end
					end

					-- Add any new ids from this array diff.
					for _, docId in ipairs( newDocIdsArray ) do
						if not unfetchedDocIdsSet[ docId ] then
							newDocIdsToFetchArray[ #newDocIdsToFetchArray + 1 ] = docId
						end
					end

					-- Add the newDocIds to the set of currently watched docIds
					-- to be used for the ongoing watching of the docIds after
					-- their current values have been retrieved.
					currWatchedDocIds = 
						WFFunctionUtils_reduce( 
							currWatchedDocIds, addToSet, newDocIdsToFetchArray )

					docIdsBeingFetchedArray = newDocIdsToFetchArray

					local docValuesForNewIds =
							WFRxBatchedBulkOperation( "WFRxDocValueWatcher", docStore, newDocIdsToFetchArray, 200, 1, Rx.Observable.just )
								:scan( KSTableUtils_appendArray, {} )
								:takeLast( 1 )
								:flatMapLatest( 
									function( docValues )
									
										return Rx.Observable.from( docValues )
												:tapOnCompleted(
													function()
														docIdsBeingFetchedArray = {}
													end )

									end )

					local changesToAllWatchedDocIds =
							getChangesFeedForDocIds( currWatchedDocIds )

					return docStore:changeCounter()
						:take( 1 )
						:flatMapLatest( function( initialChangeCounter )

							if continueFromStream:getValue() == nil then
								continueFromStream:onNext( initialChangeCounter + 1 )
							end

							return Rx.Observable.concat(
								docValuesForNewIds,
								changesToAllWatchedDocIds )

						end )

				end )

end

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

return {

	make = make,

	reportDocValueIfWatched = reportDocValueIfWatched,

}
