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

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

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

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.core.utils
local WFArguments = require 'WFArguments'

	-- WF.core.logger
local WFLogger = require 'WFLogger'

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

local check = assert( WFArguments.check )

local logger = WFLogger( 'WFRxBatchedBulkSetDocValue' )

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

local function getBatch( t, first, last )
	
	local sub = {}
	
	for i = first, last do
		sub[ #sub + 1 ] = t[ i ]
	end
	
	return sub
	
end

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

return function( desc, docStore, docs, batchSize )

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

	local count = #docs
		
	if count < 1 then
		return Observable.just( true )
	end
	
	if batchSize and batchSize < 1 then
		batchSize = 1
	end
	
	batchSize = batchSize or 100
	local totalBatches = math.ceil( count / batchSize )
	
	local completedBatches = 0
	
	local setDocValueSubscriptions = {}
	
	local batchesCompleteStream = Rx.Subject()
	
	local function cleanup( i )
	
		local subscription = setDocValueSubscriptions[ i ]
		setDocValueSubscriptions[ i ] = nil
		
		if subscription then
			subscription:dispose()
		end
		
		completedBatches = completedBatches + 1
		
		logger:infof( "Finished batch %d of %d total batches.", completedBatches, totalBatches )
		
		if completedBatches == totalBatches then
			
			batchesCompleteStream:onNext( true )
			batchesCompleteStream:onCompleted()
			
		end
	
	end

	if count > 200 then

		-- 2018.09.04 bkruse
		-- 
		-- WARNING: despite the batched nature of this API, it will effectively
		-- schedule all the updates at once and may starve the docStore.
		-- 
		-- If you don't need atomic updates (i.e. don't need all the overlays
		-- applied immediately), consider using
		-- WFRxThrottledBatchedBulkApplyOverlay, which waits for each batch to
		-- complete before starting the next.
		-- 
		-- 
		logger:warnf( "%s is applying a large number (%s) of docstore updates in %s batches of %s.",
						count, totalBatches, batchSize, desc )

	end
	
	for i = 1, count, batchSize do 
		
		logger:infof( "Now processing docs from range %d through %d.", i, i + batchSize - 1 )
		
		local batchOfDocs = getBatch( docs, i, i + batchSize - 1 )
		
		setDocValueSubscriptions[ i ] = 
			docStore:bulkSetDocValue( batchOfDocs, nil, desc )
			
				:tapOnError( function( maybeErrorMessage )
						
					if maybeErrorMessage then
						logger:errorf( "WFRxBatchedBulkSetDocValue Failed: %s", maybeErrorMessage )
					end
		
				end )
			
				:finally( function()
					cleanup( i )
				end )
				
				:subscribe()
				
	end 

	return batchesCompleteStream

end
