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

WFCatBinaryFetcherBase.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2018 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
local WFCloseSupport = require 'WFCloseSupport'
local WFLogger = require 'WFLogger'
local KSTableUtils = require 'KSTableUtils'
local KSIdGenerator = require 'KSIdGenerator'
local WFRxContinuousTask = require 'WFRxContinuousTask'

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

local logger = WFLogger 'WFCatBinaryFetcherBase'
	--:setLevel( 'DEBUG' )

local ERROR_RETRY_DELAY = 60 * 10 -- In secs.
	-- If there are errors and there are no updates to
	-- the style table, then we need to trigger the
	-- download process to retry the error cases.

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

-- Creates a separate function for downloading of 'v2' presets and profiles.
-- Per consults with Brian K, it is probably not wise to try to channel download
-- of presets and profiles through the pixel path provider state machinery,
-- as there is a 'baked in assumption' that the downloadable assets have a
-- docValue.type of 'image'. It might be feasible (though probably overkill)
-- to implement a separate state machine for managing profile/preset sync
--
-- Another possible path, per Brian is:
--[[

Actually thinking further, I think preset sync is very similar to `RenditionPerAssetPathFetcher.lua`

`RenditionPerAssetPathFetcher.lua` ensures we have at one least one local rendition per each `assetId` in the catalog. That's pretty much the same as ensuring we have a local preset file for each `presetId` in the catalog.

Here are the `RenditionPerAssetPathFetcher.lua` implementation details:

1. it sets up a `"allLocalAssetIds"` submodel which has a `:value()` stream that emits

```{ allAssetIds, {} }
{ addToAllAssetIds, removeFromAllAssetIds }
{ addToAllAssetIds, removeFromAllAssetIds }
...
```

2. it sets up a `"idlePixelPaths"` submodel. The submodel `:sendEvent( "fetchPaths" )` call drives the current work. The submodel `:value()` stream is the ongoing results from the `:sendEvent( "fetchPaths" )` calls.

3. The `"allLocalAssetIds"` submodel `:value()` stream is fed into a "needs rendition path" queue. That queue is popped on an `repeat ... until false` `AgEventLoup` task. The inner loop of that task spends most of its time in a wait state unless no current rendition downloads are pending and the "needs rendition path" queue is non empty.

When the wait is lifted, the queue is popped and the result is passed to `:sendEvent( "fetchPaths" )` on `"idlePixelPaths"` submodel.

4. As results come back, we update the current rendition downloads are pending list as needed

So if you copy `RenditionPerAssetPathFetcher.lua` to say, `LocalPresetSyncer.lua` file, make these changes plus implementing the wf model code to fulfill them, you'll be be pretty far along with preset sync:

1. In `LocalPresetSyncer.lua`, replace the `"allLocalAssetIds"' submodel  with a new `"needsPresetSyncIds"` submodel which should emit a stream of the following

```{ needsPresetDownloadPresetIds, {} }
{ addToNeedsPresetDownloadPresetIds, removeFromNeedsPresetDownloadPresetIds }
{ addToNeedsPresetDownloadPresetIds, removeFromNeedsPresetDownloadPresetIds }
...
```

2. In the wf model code, implement `"needsPresetSyncIds"` so that it first get the list of all `presetIds` from the doc store and filters out all of them we either can't download or don't need to download. Then as updates come in (when presets are added, remove, or updated), emit the updates as arrays of adds and deletes. See `WFCatAssets.allLocalAssetIdsStream` for something very similar to that.

3. In `LocalPresetSyncer.lua`, replace `"idlePixelPaths"` submodel with `"ensurePresetIsLocal"` submodel and call `:sendEvent( "fetchPreset" )` on it.

4. In the wf model code, implement that `:sendEvent( "fetchPreset" )` to accept one `presetId` at a time. Each time the model gets a new `presetId`, cancel the current download (if there is one) and start downloading the new `presetId` after checking the local path. In any case, when done with the `presetId` return `{ presetId, isLocal, maybeErrorString }` on the `:value()` stream so that the `AgEventLoup` task will unblock its wait.

5. In `LocalPresetSyncer.lua`, make whatever additional changes needed so that `"needsPresetSyncIds"` submodel `:value()` updates the work queue and that the `"ensurePresetIsLocal"` submodel `:value()` output is handled correctly.

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

local function addToMap( t1, t2 )

	for k, v in pairs( t2 ) do 
		t1[ k ] = v
	end
end

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

local function removeFromMap( t1, t2 )

	for k, _ in pairs( t2 ) do 
		t1[ k ] = nil
	end
end

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

local function isVolumeOfflineOrOutOfSpace( self )

	return self._volumeStatus.offline or self._volumeStatus.outOfSpace

end

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

local function incrementNumOfInFlightRequests( self )

	self._numOfInFlighRequests = self._numOfInFlighRequests + 1
	self._numOfInFlightRequestsStream:onNext( self._numOfInFlighRequests )
	
	if self._test_numOfInFlightRequests then
		self._test_numOfInFlightRequests[ #self._test_numOfInFlightRequests + 1 ] = self._numOfInFlighRequests
	end
	logger:debugf( "Num of in-flight requests: %s", self._numOfInFlighRequests )
end

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

local function decrementNumOfInFlightRequests( self )

	self._numOfInFlighRequests = self._numOfInFlighRequests - 1
	self._numOfInFlightRequestsStream:onNext( self._numOfInFlighRequests )
	
	if self._test_numOfInFlightRequests then
		self._test_numOfInFlightRequests[ #self._test_numOfInFlightRequests + 1 ] = self._numOfInFlighRequests
	end
	logger:debugf( "Num of in-flight requests: %s", self._numOfInFlighRequests )
end

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

local function maybeDisposeInflightRequest( self, nextRequestKey )

	local requestInfo = self._inFlightRequests[ nextRequestKey ]
	if requestInfo and requestInfo.subcription then
		-- Count will be decremented in the
		-- finally function associated with the
		-- stream being cancelled.
		-- The inflight request entry will be
		-- removed after the dispose is complete
		-- in the finally fn of the subscription.
		
		requestInfo.subcription:dispose()
		return true
	end
	
	return false
end

--------------------------------------------------------------------------------
-- Forward declaration
local mayBeSendOneOrMore

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

local function getDownloadSubscriptionForAsset( self, requestKey, requestValue, requestId )

	return self:_doOneDownloadWithResponseStream( requestKey, requestValue, requestId )
		
		:tap( function( requestKey, success, pathOrErr )
			
			if success then
				logger:infof( "Downloaded style asset %s : path=%s", requestKey, pathOrErr )
				
				-- Remove it from the errorStyles.
				self._errorStyles[ requestKey ] = nil
			else
						
				logger:errorf( "Error for Asset %s = %s", requestKey, pathOrErr )
				self._errorStyles[ requestKey ] = requestValue

			end
		end )
		
		:finally( function()
			
			-- Finally gets called even if the subscription
			-- is disposed.
			
			-- Count has to be decremented irrespective of
			-- whether the request is current, since we are
			-- expecting a response for each and every
			-- request, except for the requests which got
			-- cleared as part of stopping the continuous
			-- process. 
			-- This is needed so that we dont decrement
			-- the count, after the state is cleared.
			
			if self._inFlightRequestIds[ requestId ] then
				decrementNumOfInFlightRequests( self )
				self._inFlightRequestIds[ requestId ] = nil
			end
			
			-- Remove the asset from inflight request list,
			-- if we are processing the response for the
			-- latest request for that asset. 
			local requestInfo = self._inFlightRequests[ requestKey ]
			
			if requestInfo and requestInfo.requestId == requestId then
				
				-- If a request for assetId is disposed
				-- and then raised again, then it is possible
				-- that the inFlight list is updated with the
				-- new entry. Hence remove it only, if the
				-- requestId matches.
				
				self._inFlightRequests[ requestKey ] = nil
			end
			
			mayBeSendOneOrMore( self )
		end )
		
		:subscribe()
end
--------------------------------------------------------------------------------

mayBeSendOneOrMore = function( self )

	if isVolumeOfflineOrOutOfSpace( self ) then
		-- We cannot download since volume is
		-- not servicable. 
		return
	end
	
	if not self._isStarted then
		-- The continuous task is not running now.
		return
	end
	
	while( self._numOfInFlighRequests < self._maxNumOfFlightRequests 
		and next( self._pendingStyles ) ) do
		
		local nextRequestKey, nextRequestValue = next( self._pendingStyles )
		
		-- Remove from the pending list and add to
		-- the inFlightList. 
		-- We are ensuring, at any point of time, 
		-- there are only _maxNumOfFlightRequests
		-- by checking the inFlightRequest count.
		
		self._pendingStyles[ nextRequestKey ] = nil
		
		-- If a request for this asset is already
		-- in progress, need to cancel it and
		-- add a new request.
		-- Count will be decremented in the
		-- finally function associated with the
		-- stream being cancelled.
		
		maybeDisposeInflightRequest( self, nextRequestKey )
		
		-- We are making sure that the subscribe and
		-- dispose is always called in order, by
		-- making both of them async. But if we
		-- call dispose and then subscribe in quick
		-- succession, then it is possible that
		-- the response of disposing off the subscription
		-- comes after the call to subscribe. Hence
		-- we need to make sure that we are updating
		-- the proper inflight request. For that we
		-- are using the requestId.
		
		-- Here the previous subscription, if any, is
		-- disposed and the new one is stored. I believe,
		-- in all cases, the old subscrition will be
		-- held untill it is disposed off and we will
		-- receive a call in finally. 
		local requestId = KSIdGenerator.uuidInLowercaseHex()
		
		self._inFlightRequestIds[ requestId ] = true
		
		self._inFlightRequests[ nextRequestKey ] = {
			subcription = getDownloadSubscriptionForAsset( self, 
				nextRequestKey, nextRequestValue, requestId ),
			requestId = requestId,
			requestValue = nextRequestValue,
		}
		incrementNumOfInFlightRequests( self )
		
	end
end

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

local function cancelStaleRequests( self, assetIds )

	for i, assetId in ipairs( assetIds ) do
		
		maybeDisposeInflightRequest( self, assetId )
		
	end
end

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

local function mayBeRetryErrorRecords( self )

	return Rx.Observable.just()
		:map( function()
			
			if not next( self._errorStyles ) then
				-- No error records to process.
				return
			end
			
			if next( self._pendingStyles ) then
				-- There are pending records to process
				-- Lets not process error records in 
				-- this cycle.
				return
			end
			
			if next( self._inFlightRequests ) then
				-- Assume the download is taking more
				-- time as compared to the error
				-- interval. There is a chance that
				-- the same request will get added
				-- to the queue again. Avoiding this
				-- raise condition by waiting for all 
				-- the requests to finish
				return
			end
			
			-- Retry all error cases as if those
			-- are the normal pending ones.
			addToMap( self._pendingStyles, self._errorStyles )
			self._errorStyles = {}
			mayBeSendOneOrMore( self )
		end )
		
		:catch( function( errorMessage )
		
			return self._continuousTaskForErrorRetry:throwIfNotStopped( errorMessage )
		end )
	
end


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

local function resetCurrentDownloadStates( self )

	self._inFlightRequests = {}
	self._inFlightRequestIds = {}
	self._pendingStyles = {}
	self._errorStyles = {}
	self._numOfInFlighRequests = 0
end

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

local function disposeAllDownloads( self )

	for assetId, requestInfo in pairs( self._inFlightRequests or {} ) do
		if requestInfo.subscription then
			requestInfo.subcription:dispose()
		end
	end
	
end

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

local WFCatBinaryFetcherBase = {
	
	_catalog = false,
	
	_binaryDownloader = false,
	
	_stdProfileDownloader = false,
	
	_downloadStreamSubscription = false,
	
	_downloadResponseStream = false,
	
	_inFlightRequests = false,
	
	_inFlightRequestIds = false,
	
	_numOfInFlighRequests = false,
	
	_numOfInFlightRequestsStream = false,
	
	_pendingStyles = false,
	
	_errorStyles = false,
	
	_continuousTaskForErrorRetry = false,
	
	_maxNumOfFlightRequests = false,
	
	_test_numOfInFlightRequests = false,
	
	_volumeGuid = false,
	
	_volumeStatus = false,
	
	_isStarted = false,
	
	WFCloseSupport.addCloseSupportMethods,
}

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

function WFCatBinaryFetcherBase:__init( catalog, maxFlightRequests )

	self._catalog = catalog
	self._volumeGuid = catalog:fileSpec():pixelsPath().volumeGuid
	self._downloadResponseStream = Rx.Subject()
	self._maxNumOfFlightRequests = maxFlightRequests
	self._numOfInFlightRequestsStream = Rx.BehaviorSubject( 0 )

	-- For error retries
	local errorRetryTaskDelays = {
		progressiveBackoff = false,
		errorDelay = ERROR_RETRY_DELAY,
		maxDelay = ERROR_RETRY_DELAY,
	}

	local continuousTaskForErrorRetry = function() return mayBeRetryErrorRecords( self ) end

	self._continuousTaskForErrorRetry = WFRxContinuousTask( "StyleErrorRetryTask", continuousTaskForErrorRetry, errorRetryTaskDelays )
	
	self._volumeStatus = {}
end

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

function WFCatBinaryFetcherBase:getResponseStream()
	
	return self._downloadResponseStream
end

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

function WFCatBinaryFetcherBase:start()
	
	self._isStarted = true
	resetCurrentDownloadStates( self )
	self._continuousTaskForErrorRetry:startContinuousTask()
end

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

function WFCatBinaryFetcherBase:stop()

	self._isStarted = false
	self._continuousTaskForErrorRetry:stopContinuousTask()
	
	disposeAllDownloads( self )
	resetCurrentDownloadStates( self )
end

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

function WFCatBinaryFetcherBase:reportStatusOfVolumes( statusOfVolumes )

	local offline = statusOfVolumes.offlineVolumes[ self._volumeGuid ]
	local outOfSpace = statusOfVolumes.volumesOutOfSpace[ self._volumeGuid ]
	local overIdleSpaceThreshold = statusOfVolumes.volumesOverIdleSpaceThreshold[ self._volumeGuid ]
	
	local newVolumeStatus = {
		volumeGuid = self._volumeGuid,
		offline = offline,
		outOfSpace = outOfSpace,
		overIdleSpaceThreshold = overIdleSpaceThreshold
	}
	
	if KSTableUtils.deepEquals( self._volumeStatus, newVolumeStatus ) then
		-- No change in status
		return
	end
	
	self._volumeStatus = newVolumeStatus
	
	-- Currently we are not cancelling the
	-- inFlight requests when volume is down.
	-- We will just stop any new requests.
	
	if self._isStarted then
		-- If we had pending requests before,
		-- we will need to add it for processing
		mayBeSendOneOrMore( self )
	end
end

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

function WFCatBinaryFetcherBase:close()
	
	self:stop()
end

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

function WFCatBinaryFetcherBase:isDownloading()
	
	return self._numOfInFlightRequestsStream
		:distinctUntilChanged()
		:map( function( num )
			return num > 0
		end )
end

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

function WFCatBinaryFetcherBase:addAndRemoveRequests( additions, removals )
	
	if additions then
		-- Update the pending list with
		-- the new items.
		addToMap( self._pendingStyles, additions )
	end
	
	if removals then
		removeFromMap( self._pendingStyles, removals )
		
		-- Remove the deleted entries from the
		-- error list as well
		removeFromMap( self._errorStyles, removals )
		
		-- Cancel in-flight requests, if any
		-- for the removed assets.
		cancelStaleRequests( self, removals )
	end
	
	-- Maybe some new styles have got added.
	-- Trigger the processing chain.
	mayBeSendOneOrMore( self )
end

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

function WFCatBinaryFetcherBase:maybeChangePriorityForRequests( requests, newHttpPriority )
	
	if not self._isStarted then
		return
	end
	
	if not requests then
		return
	end
	
	for k, requestValue in pairs( requests ) do
		
		-- Change priority in pending requests if present
		if self._pendingStyles[ k ] then
			self._pendingStyles[ k ].httpPriority = newHttpPriority
		end
		
		-- Change priority in error requests if present
		if self._errorStyles[ k ] then
			self._errorStyles[ k ].httpPriority = newHttpPriority
		end
		
		-- Update priority if already is in-flight
		-- Dispose and readd
		if maybeDisposeInflightRequest( self, k ) then
			requestValue.httpPriority = newHttpPriority
			self:addAndRemoveRequests( { [ k ] = requestValue } )
		end
	end
	
	-- Trigger the processing chain.
	mayBeSendOneOrMore( self )
end


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

function WFCatBinaryFetcherBase:_doOneDownload( requestKey, requestValue, requestId )

	logger:errorf( "To be implemented by overriding class" )
	return Observable.throw( "Abstract method" )
end

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

function WFCatBinaryFetcherBase:_doOneDownloadWithResponseStream( requestKey, requestValue, requestId )

	return self:_doOneDownload( requestKey, requestValue, requestId )
		
		:catch( function( err )
		
			return Observable.just( requestKey, false, err )
		end )
		
		:tap( function( requestKey, success, pathOrErr )
			
			logger:infof( "Style asset download %s, Status:%s, pathOrErr=%s", requestKey, success, pathOrErr )
			self._downloadResponseStream:onNext( success, requestKey, pathOrErr )
			
		end )
end

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

function WFCatBinaryFetcherBase:setTestModeAndParams()
	
	self._test_numOfInFlightRequests = {}
end

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

-- To be used only by tests.
function WFCatBinaryFetcherBase:_test_getNumOfInFlightRequestArray()
	
	return self._test_numOfInFlightRequests
end

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

return Rx.Class 'WFCatBinaryFetcherBase'( WFCatBinaryFetcherBase )