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

WFRxBatchedBulkOperation.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2019 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.

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

	-- RxLua
local Rx = require 'Rx'

local WFArguments = require 'WFArguments'

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

local check = assert( WFArguments.check )

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

local function batchedDocIdGenerator( docIds, batchSize )

	local curOffset = 1
	local numDocs = #docIds

	local batchedDocIds = Rx.ReplaySubject( 1 )

	local function sendNextBatch()

		if curOffset <= numDocs then

			local nextOffset = math.min( curOffset + batchSize, numDocs + 1 )

			local batchOfIds = {}
			for i = curOffset, nextOffset - 1 do
				batchOfIds[ #batchOfIds + 1 ] = docIds[ i ]
			end

			batchedDocIds:onNext( batchOfIds )

			curOffset = nextOffset

		else

			batchedDocIds:onCompleted()

		end

	end

	return batchedDocIds, sendNextBatch

end

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

--[[!

	\brief A function used to get and process docValues in batches. This function
		will only request maxSimultaneous*batchSize doc values regardless of how
		long the operations take to complete.

	\param desc (string) description of operation to use for debugging and diagnostic logs
	\param docStore (docstore) A valid WFRxDocStore.
	\param docIds (table) A table of doc ids.
		Optional arguments to this api are passed as named parameters in the
		docIds table they are:
		* \c label (optional, string): Specifies what label-ed revision to
			get for the documents.
		* \c includeInitialChangeCounter (optional, boolean): if true, return the first
			changeCounter (sequence number) related to the documents.
	\param batchSize (number) The batch size, the number of doc
		values to fetch for each call to bulkDocValue.
	\param maxSimultaneous (number) The maximum number of batches to run in parallel.
	\param observableFn (function) The operation to be performed, this function
		is passed in an array of docValues and must return an observable that will
		perform the operation on those docValues.
		Example:
			local function observableFn( batchOfDocValues )
				local overlays = {}
				for i, docValue in ipairs( batchOfDocValues ) do
					overlays[ i ] = {
						_id = docValue._id,
						_localOnly = { processed = true },
					}
				end
				return docStore:bulkApplyOverlay( overlays )
			end

	\return <code>Observable</code> An observable that runs observableFn on batches of
		of doc values, as they're available. The output of the observableFn is the
		stream of values it sends.

--]]

return function( desc, docStore, docIds, batchSize, maxSimultaneous, observableFn )

	check( 'string', 'WFRxDocStore', 'table', 'number', 'number', 'function' )

	local batchedDocIds, sendNextBatch = batchedDocIdGenerator( docIds, batchSize )

	sendNextBatch()

	return batchedDocIds:map( function( batchOfIds )

		return Rx.Observable.defer( function()

			sendNextBatch()

			-- Pass along the optional arguments to bulkDocValue.

			batchOfIds.label = docIds.label
			batchOfIds.includeInitialChangeCounter = docIds.includeInitialChangeCounter

			return docStore:bulkDocValue( batchOfIds, nil, desc )
			:flatMap( function( batchOfDocValues )
				return observableFn( batchOfDocValues )
			end )

		end )

	end )

	:flatMerge( maxSimultaneous )

end