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

WFRxThrottledBatchedApplyOverlayToIds.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2020 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 Observable = Rx.Observable

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

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

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

local check = assert( WFArguments.check )

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

local WFFunctionUtils_reduce = WFFunctionUtils.reduce

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

local function addToChunkedArray( chunks, val, chunkSize )

	local currChunk = chunks[ #chunks ]

	local currChunkSize = #currChunk
	
	if currChunkSize > 0 and currChunkSize  % chunkSize == 0 then

		currChunk = {}

		chunks[ #chunks + 1 ] = currChunk

	end
		
	currChunk[ #currChunk + 1 ] = val

	return chunks

end

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

local function chunkedArray( arr, chunkSize )

	return WFFunctionUtils_reduce( { {} }, addToChunkedArray, arr, chunkSize )

end

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

local _lastBulkApplyOverlayDeliveryTimes = KSTableUtils.makeWithMode( "kv" )

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

--[[!

	\brief A function used to batch/throttle calls to the ApplyOverlayToIds doc 
		store API.
	
	\param desc (string) description of operation to use for debugging and diagnostic logs
	\param docStore (docstore) A valid WFRxDocStore.
	\param overlay (table) An Oz document overlay.
	\param docIds (table) A table of local or full document IDs.
	\param batchSize (number) The batch size. AKA the number of overlays
		sent on each call to bulkApplyOverlay.
	\param callsPerSecondLimit (number) The max number of API
		calls that we'll make to the doc store per second.
	
	\return <code>Observable</code> An observable that completes when all of the
		overlays have been applied.
		
--]]

return function( desc, docStore, overlay, docIds, batchSize, callsPerSecondLimit )

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

	local count = #docIds

	if count < 1 then
		return Observable.just( true )
	end

	if count <= batchSize then
		return docStore:applyOverlayToIds( overlay, docIds, desc )
	end
	
	if batchSize and batchSize < 1 then
		batchSize = 1
	end
	
	local minWaitUpperBound = 1 / callsPerSecondLimit

	local scheduler = WFRxScheduler.sharedScheduler()

	local chunkedIds = chunkedArray( docIds, batchSize )

	return Rx.Observable.from( chunkedIds )

		:map( function( batchOfIds )

			return Rx.Observable.defer( function()

				local lastDeliveredTime =
						_lastBulkApplyOverlayDeliveryTimes[ docStore ]
					
				local delayInSecs =
						not lastDeliveredTime and
							0 or
							math.max(
								minWaitUpperBound -
									( KSDate.timeIntervalSinceAppLaunch() -
										lastDeliveredTime ),
								0 )

				return docStore 
					:applyOverlayToIds( overlay, batchOfIds, desc )
					:delay( delayInSecs, scheduler )
					:tap( function()

						_lastBulkApplyOverlayDeliveryTimes[ docStore ] =
							KSDate.timeIntervalSinceAppLaunch()

					end )

			end )

		end )
		
		:flatMerge( 1 )

		:concat( Observable.just( true ) )
		
		:takeLast( 1 )

end
