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

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

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

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

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

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

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

	-- WF.core.os
local KSDate = require 'KSDate'

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

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFOzDocUtils = require 'WFOzDocUtils'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFRingBuffer = require 'WFRingBuffer'
local WFURI = require 'WFURI' 

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'

	-- WF.oz_client.managed_catalog
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatTrash = require 'WFCatTrash'
local WFCatUrlUtils = require 'WFCatUrlUtils'

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

local resolveToAbsoluteUrl = assert( WFCatUrlUtils.resolveToAbsoluteUrl )

local isAssetRevisionOnOz = assert( WFCatIndexUtils.isAssetRevisionOnOz )
local isAssetDocument = assert( WFCatIndexUtils.isAssetDocument )
local isTrashAssetDocument = assert( WFCatIndexUtils.isTrashAssetDocument )
local isValidAssetDocument = assert( WFCatIndexUtils.isValidAssetDocument )

local WFCatTrash_copyLocalFilesOverlayForUnmodifiedAndFullyUploadedDoc = assert( WFCatTrash.copyLocalFilesOverlayForUnmodifiedAndFullyUploadedDoc )
local WFCatAssetUtils_binaryUploadPending = assert( WFCatAssetUtils.binaryUploadPending )
local WFCatAssetUtils_hasLocalOriginalOrProxy = assert( WFCatAssetUtils.hasLocalOriginalOrProxy )
local WFCatAssetUtils_getLocalExternalDevelopSettingsPath = assert( WFCatAssetUtils.getLocalExternalDevelopSettingsPath )

local check = assert( WFArguments.check )

local runUntilDone = WFRxRunUntilDone

local timeNow = assert( KSDate.timeNow )
local timeNowForAnimation = assert( KSDate.timeNowForAnimation )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )
local KSTableUtils_setNested = assert( KSTableUtils.setNested )

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

local function getNextUpdateFeedLinkLocalDocId( isCatalogPuller, customId )

	if isCatalogPuller then
	
		return 'Oz-nextUpdateFeedLink'
		
	else
	
		customId = string.gsub( customId, "%s+", "_" )
			-- Replace white space with underscores since doc IDs cannot
			-- contain spaces.
			
		return string.format( "Oz-nextUpdateFeedLink--%s", customId )
		
	end

end

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

local function getIsDoingInitialSyncLocalDocId( isCatalogPuller, customId )

	if isCatalogPuller then
	
		return 'Oz-isDoingInitialSync'
		
	else
	
		customId = string.gsub( customId, "%s+", "_" )
			-- Replace white space with underscores since doc IDs cannot
			-- contain spaces.
			
		return string.format( "Oz-isDoingInitialSync--%s", customId )
		
	end
	
end

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

local function remapOzDocumentData( self, ozDocument, baseUrl )

	local result = WFOzDocUtils.remapBasicOzDocument( ozDocument, baseUrl )
	
	-- Remap deletion to our flavor of deletion.
	if ozDocument.deleted then

		ozDocument.deleted = nil
		result._deleted = true

		for k in pairs( result ) do
			if not k:match( '^_' ) and k ~= 'type' and k ~= 'subtype' then
				result[ k ] = nil
			end
		end

	end
	
	-- In case Oz fails to send revision_ids, squawk about it, but force
	-- push the doc anyway.
	if result and ( not result._revIds or #result._revIds == 0 ) then

		-- No need to squawk if it's a deletion. No revision_ids expected.
		
		if not result._deleted then
			self._logger:infof( "%s: Oz did not provide revision_ids in this document: %s", self._description, result )
		end
		
		result._revIds = { KSIdGenerator.uuidInLowercaseHex(), force = true }

	end
	
	local assetTypeHandler = self._assetTypeHandler[ result.type ] 

	if assetTypeHandler then
		assetTypeHandler( self, result )
		
		-- Assert _localOnly as it would overwrite existing data in the DocStore
		-- due to the Puller using :setDocValue()
		
		assert( result._localOnly == nil, 
			'_localOnly is not permitted here, as it would overwrite existing data' )
			
	end

	local spaceId = self._spaceId
	
	if spaceId then
		result._spaceId = spaceId
	end

	return result

end

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

local function convertUpdatesToDocValues( self, updates, baseUrl )

	if type( updates ) ~= 'table' then
		return updates
	end

	-- Read an Oz updates feed and convert to input for :bulkSetDocValue.

	local result = { label = 'oz' }
	
	local remappedDocCount = 0
	
	for _, updateDoc in ipairs( updates ) do

		self._logger:debugf( "%s: about to remap doc from Oz: %s", self._description, updateDoc )

		local remappedDoc = remapOzDocumentData( self, updateDoc, baseUrl )
		
		if remappedDoc then
		
			remappedDocCount = remappedDocCount + 1
			
			result[ remappedDocCount ] = remappedDoc

			self._logger:infof( "%s: remapped doc: %s\n\n", self._description, remappedDoc )

			local docType = remappedDoc.type

			if docType then
			
				local subtype = remappedDoc.subtype or 'NULL'
			
				local function updateDocumentCount( typeOfUpdate )
				
					local total = KSTableUtils_derefOpt( self._documentsReceivedByType, typeOfUpdate, 'total' ) or 0

					KSTableUtils_setNested( self._documentsReceivedByType, total + 1, typeOfUpdate, 'total' )
				
					local docCount = KSTableUtils_derefOpt( self._documentsReceivedByType, typeOfUpdate, docType, subtype ) or 0

					KSTableUtils_setNested( self._documentsReceivedByType, docCount + 1, typeOfUpdate, docType, subtype )
				
					return self._documentsReceivedByType
				
				end
			
				local typeOfUpdate
				
				if remappedDoc._deleted then
					typeOfUpdate = 'delete'
				else
					typeOfUpdate = 'update'
				end
					
				self._documentsReceivedByType = updateDocumentCount( typeOfUpdate )
					
				if self._recentlyPulledDocIds then
				
					WFRingBuffer.addItem( self._recentlyPulledDocIds, remappedDoc._id )
					
				end
					
				if self._documentsReceivedByTypeStream then
					self._documentsReceivedByTypeStream:onNext( self._documentsReceivedByType )
				end
				
			end
		
		end

	end

	return result

end

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

-- Needed to avoid storing the 'force_upgrade_error' parameter, if present.

local function stripForcedUpgrade( str )

	local pattern = "force_upgrade_error=1[&]*"

	return str:gsub( pattern, "" ):gsub( "%?$", "" ):gsub( "&$", "" )

end

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

local function updateLocalFeedLink( self, newLink )

	if self._localNextUpdateFeedLink then
		self._localNextUpdateFeedLink:onNext( newLink )
	end
	
end

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

local function checkForInitialSync( self, docStore )
	
	if not self._hasCheckedForInitialSync then 
		
		self._hasCheckedForInitialSync = true
		
		return docStore:localDocValue( self._isDoingInitialSyncLocalDocId )
			:map( function( docValue )
				
				if docValue.isDoingInitialSync == nil then 
					docStore:setLocalDocValue {
						_id = self._isDoingInitialSyncLocalDocId,
						isDoingInitialSync = true,
						beginTimestamp = timeNow(),
					}
					:call( runUntilDone )
				end
				
				-- if the isDoingInitialSync value hadn't yet been set, consider it true
				self._isDoingInitialSync = docValue.isDoingInitialSync == nil or docValue.isDoingInitialSync
				
				return self._isDoingInitialSync
			end )
			
	else 
		
		return Observable.just( self._isDoingInitialSync )
		
	end
end

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

local function markInitialSyncCompleteIfNecessary( self, docStore )
	
	if self._isDoingInitialSync then 
		self._isDoingInitialSync = false
		
		docStore:applyOverlayLocalDocValue {
			_id = self._isDoingInitialSyncLocalDocId,
			isDoingInitialSync = false,
			endTimestamp = timeNow(),
		}
		:call( runUntilDone )
	
	end
end 

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

local function retryDelayRequested( headers )
	return headers and headers.status_code == '200' and tonumber( headers[ 'Retry-After' ] )
end

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

local function maybeAddDelayFromResponseHeader( headers )

	local retryDelay = retryDelayRequested( headers )

	if retryDelay then
		return Observable.timer( retryDelay, WFRxScheduler.sharedScheduler() )
	end

	return Observable.just()

end

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

local function maybeSetRetryDelayRequestByOz( self, headers )

	local retryDelay = retryDelayRequested( headers )

	if retryDelay then

		if self._retryDelayRequestByOz then
			-- Don't think this can happen.
			self._logger:errorf( "%s: Trying to add a retryDelay when one is already set, using new delay", self._description )
		end

		self._retryDelayRequestByOz = retryDelay
		self._retryDelayRequestByOzAtTime = timeNowForAnimation()

	end

end

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

local function maybeAddDelayRequestByOz( self )

	if self._retryDelayRequestByOz then

		local timeSinceDelayRequested = timeNowForAnimation() - self._retryDelayRequestByOzAtTime

		local retryDelay = self._retryDelayRequestByOz - timeSinceDelayRequested

		self._retryDelayRequestByOz = nil
		self._retryDelayRequestByOzAtTime = nil

		if retryDelay > 0 then
			return Observable.timer( retryDelay, WFRxScheduler.sharedScheduler() )
		end

	end

	return Observable.just( true )

end

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

local function prefetchAlbumHierarchyImp( self, albumsLink, iterationCount )
	
	if not albumsLink or iterationCount > 20 then
		return Observable.just()
	end
	
	return self._catalogSpec:get( albumsLink, {
			returnHttpErrorsAsData = true
		} )
		:map( function( albumsJson, headers )
			
			if not headers or headers.status_code ~= '200' then
				return Observable.just()
			end
			
			-- decode the albums json into docs and splat into the doc store
			local decodedAlbumJson = KSJsonDecoder.decodeWebSafeJson_stripNull( albumsJson )
						
			local ozAlbums = KSTableUtils_derefOpt( decodedAlbumJson, "resources" )
			local baseUrl = KSTableUtils_derefOpt( decodedAlbumJson, 'base' )
			local albumDocs = convertUpdatesToDocValues( self, ozAlbums, baseUrl )
			
			local maybeNextLink = KSTableUtils_derefOpt( decodedAlbumJson, 'links', 'next', 'href' )
			
			if maybeNextLink then
				maybeNextLink = WFURI.resolve( baseUrl, maybeNextLink )
			end
						
			return albumDocs, maybeNextLink, headers
			
		end )
		:flatMap( function( albumDocs, maybeNextLink, headers )
		
			if albumDocs and #albumDocs > 0 then
				return self._catalog:docStore():bulkSetDocValue( albumDocs )
					:map( function()
						return maybeNextLink, headers
					end )
			else
				return Observable.just()
			end
		
		end )
		:flatMap( function( maybeNextLink, maybeHeaders )
			
			if not maybeNextLink then
				return Observable.just()
			end
			
			return prefetchAlbumHierarchyImp( self, maybeNextLink, iterationCount + 1 )
				:delay( function()
					return maybeAddDelayFromResponseHeader( maybeHeaders )
				end )
		
		end )
		:catch( Observable.just() )
		
end

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

local function prefetchAlbumHierarchy( self )
	
	local catSpec = self._catalogSpec
	
	local albumsHref = string.format( '%s/albums?subtype=collection%%3Bcollection_set', catSpec:catalogHref() )
	
	return prefetchAlbumHierarchyImp( self, albumsHref, 0 )
		:takeLast( 1 )
end

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

local function remapCatalogDocument( catalogDoc )
	
	catalogDoc = WFOzDocUtils.remapBasicOzDocument( catalogDoc, catalogDoc.base )
	
	-- Map as if it came from oz as part of sync feed
	catalogDoc._labels = { oz = true }
	
	return catalogDoc
	
end

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

local function getCatalogDocFromOzAndRemap( catalog )
	
	local catalogSpec = catalog:ozCatalogSpec()
	
	return catalogSpec:get( catalogSpec:catalogHref() )
		
		:map( function( catalogJson, headers )
			
			if not headers or headers.status_code ~= '200' then
				error( "Request failed" )
			end
			
			local catalogDoc = KSJsonDecoder.decodeWebSafeJson_stripNull( catalogJson )
			catalogDoc = remapCatalogDocument( catalogDoc )
			return catalogDoc
		end )
end

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

local function ensureCatalogDocIsAvailableLocally( self )

	if self._hasEnsuredCatalogExists or not self._isCatalogPuller then
		return Rx.Observable.just()
	end

	return self._catalog:docStore():docValue( self._catalog:ozCatalogId() )
	
		:flatMap( function( docValue )
			
			if docValue and not docValue._empty then
				-- Catalog doc is present. Return.
				return Rx.Observable.just()
			end
			
			return getCatalogDocFromOzAndRemap( self._catalog )
				:flatMap( function( catalogDoc )
					return self._catalog:docStore():setDocValue( catalogDoc )
				end )
			
		end )
		
		:tap( function()
			
			-- Turn on the variable so that we dont do this task again.
			
			self._hasEnsuredCatalogExists = true
			
		end )
		
		:tapOnError( function( err )
		
			self._logger:errorf( "%s: Adding catalog document failed. Error:%s", self._description, err )
			
		end )

end

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

local function getNewestId( mapIdToNewId, id )

	local newId = mapIdToNewId[ id ]
	local numMoves = 0
	repeat
		numMoves = numMoves + 1
		local newerId = mapIdToNewId[ newId ]
		newId = newerId or newId
	until newerId == nil

	return newId, numMoves

end

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

local function bulkSetDocValuesPreservingModificationsToDeletedItemsMovedToNewIds( catalog, wfUpdateDocs, maybeLocalDocs )

	local docStore = catalog:docStore()

	-- Items with a new_id may be a deleted asset or a deleted trash asset
	-- that is being restored. In either case make sure to copy the latest
	-- revision of the asset to the new_id before actually deleting the item
	-- if the revision is not on oz yet. This handles the case where an asset
	-- was modified but deleted by a different client before we were able to
	-- sync the changes up to oz.

	local idOfAssetsBeingMoved = {}
	local mapIdToNewId = {}
	local isNewId = {}

	for i, updateDoc in ipairs( wfUpdateDocs ) do

		if updateDoc._deleted and updateDoc.type == 'asset' then

			local new_id = KSTableUtils_derefOpt( updateDoc, '_serverProvidedMetadata', 'new_id' )
			if new_id then
				local movedItemIndex = #idOfAssetsBeingMoved + 1
				idOfAssetsBeingMoved[ movedItemIndex ] = updateDoc._id
				mapIdToNewId[ updateDoc._id ] = new_id
				isNewId[ new_id ] = true
			end

		end

	end

	if #idOfAssetsBeingMoved > 0 then

		-- Eliminate any id's that were listed as a new_id in the update as they are intermediate moves.
		for i = #idOfAssetsBeingMoved, 1, -1 do
			if isNewId[ idOfAssetsBeingMoved[ i ] ] then
				table.remove( idOfAssetsBeingMoved, i )
			end
		end

		return docStore:bulkDocValue( idOfAssetsBeingMoved, nil, "MetadataPuller_getDeletedInfo" )

		:map( function( batchOfDocValues )

			local modifiedAssetOverlays = {}

			for i, docValue in ipairs( batchOfDocValues ) do

				-- If the asset hasn't finished uploading preserve it all
				-- as the asset was incomplete and can't be modified.

				-- https://jira.corp.adobe.com/browse/WF-1802
				-- We need to check isAssetDocument( docValue ) first to fix the scenaio:
				-- If doc is created in the middle of transition, it might not have type and subtype, 
				-- only with _localOnly section from original docValue
				
				if isAssetDocument( docValue ) 
					and ( not isAssetRevisionOnOz( docValue, docValue._revId  )
					or WFCatAssetUtils_binaryUploadPending( docValue ) ) then

					local modifiedDocValue
					-- Pass nil for timeOfDelete/timeOfRestore as we don't know it and it should be applied
					-- by the sync feed.

					if docValue._deleted or docValue._empty then

						-- Document was already deleted locally or didn't exist.
						-- So there will are no modifications to copy from it.

					else

						local newestId, numMoves = getNewestId( mapIdToNewId, docValue._id )

						-- Since this delete is coming from oz with a new_id we know that new_id exists
						-- on oz so we can pass docIsOnOz = true in the two calls below.
						local docIsOnOz = true

						local currentDocIsTrashAsset = isTrashAssetDocument( docValue )
						local evenNumberOfMoves = numMoves % 2 == 0

						local makeTrashAsset = ( currentDocIsTrashAsset and evenNumberOfMoves )
							or not ( currentDocIsTrashAsset or evenNumberOfMoves )

						if makeTrashAsset then
							modifiedDocValue = WFCatTrash.createTrashAssetFromDocValue( docValue, docIsOnOz, nil, nil, nil)
							modifiedDocValue.original = nil
						else
							-- Scenario:
							-- User modified an asset and deleted it, then synced.  The sync
							-- only got the delete pushed up to oz but the pull didn't occur yet
							-- The pull would result in a conflict which would cause the update
							-- to be sent.
							-- Let restored asset docValue from Oz to update correct timeOfRestore
							modifiedDocValue = WFCatTrash.createRestoredAssetFromDocValue( docValue, docIsOnOz, nil )
						end

						modifiedDocValue._id = newestId

						modifiedAssetOverlays[ #modifiedAssetOverlays + 1 ] = modifiedDocValue

					end

				elseif WFCatAssetUtils_hasLocalOriginalOrProxy( docValue ) or
					 WFCatAssetUtils_getLocalExternalDevelopSettingsPath( docValue ) then

					-- Just copy the local files section as we don't want to overwrite
					-- any modifications done by other clients.

					local newestId = getNewestId( mapIdToNewId, docValue._id )
					local filesOverlay = WFCatTrash_copyLocalFilesOverlayForUnmodifiedAndFullyUploadedDoc( docValue, newestId )
					modifiedAssetOverlays[ #modifiedAssetOverlays + 1 ] = filesOverlay

				end

			end

			return modifiedAssetOverlays

		end )

		:flatMap( function( modifiedAssetOverlays )

			return Observable.combineLatest(
				docStore:bulkSetDocValue( wfUpdateDocs, maybeLocalDocs, "MetadataPuller_new_id" ),
				Observable.just( modifiedAssetOverlays ) )

		end )

		:flatMap( function( bulkSetDocValueResults, modifiedAssetOverlays )

			if #modifiedAssetOverlays == 0 then
				return Observable.just( bulkSetDocValueResults )
			end

			-- Apply the modifications after the setDocValue in case the
			-- trash asset is in wfUpdateDocs.
			return docStore:bulkApplyOverlay( modifiedAssetOverlays, "MetadataPuller_preserveDeletedMods" )

			:map( function()
				return bulkSetDocValueResults
			end )

		end )

	end

	-- No deleted items being moved, just set the doc values.

	return docStore:bulkSetDocValue( wfUpdateDocs, maybeLocalDocs, "MetadataPuller" )

end

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

local function getLatestUpdates( self, onlyGetUpdatesOnce, numRequiredUpdates )

	check( 'WFCatMetadataPuller' )

	-- TODO [scouten 2015-07-20]: Implement connection status reporting.
	
	if self._gettingLatestUpdates then
		return self._gettingLatestUpdates
	end
	
	if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then
		-- SPAM [matthewj 2016-08-26]: If we are shutting down and the continuous task runs
		-- we can stop it from sloppily attempting additional work by returning an Observable
		-- that won't complete 
		return Observable.never()	
	end
	
	local docStore = assert( self._catalog:docStore() )
	
	-- Find out where we left off before.
	self._gettingLatestUpdates = checkForInitialSync( self, docStore )
		:flatMap( function( isInitialSync ) 
			
			return docStore:localDocValue( self._nextUpdateFeedLinkLocalDocId )
				:map( function( docValue )
					local maybeUpdatesUrl = KSTableUtils_derefOpt( docValue, 'link' )
					return isInitialSync, maybeUpdatesUrl
				end )
			
		end )
		:flatMap( function( isInitialSync, maybeUpdatesUrl )

			local result = Observable.just( maybeUpdatesUrl )

			if ( not isInitialSync ) or maybeUpdatesUrl then
				return result
			end
			
			if not self._isCatalogPuller then
				-- We only need to prefetch the album hierarchy for the catalog puller.
				return result
			end
			
			-- first time sync for the catalog, let's prime the pump with some albums
			return prefetchAlbumHierarchy( self )
				:map( function()
					return maybeUpdatesUrl
				end )
		
		end )
		:flatMap( function( maybeUpdatesUrl )

			-- Store the catalog document, if it is not
			-- preset already
			return ensureCatalogDocIsAvailableLocally( self )
				:map( function()
					return maybeUpdatesUrl
				end )
		
		end )
		:map( function( maybeUpdatesUrl )

			local updateUrl = maybeUpdatesUrl

			if not updateUrl then
			
				if self._updatesFeedUrl then
				
					-- Use the update URL provided when this puller was initiated.
				
					updateUrl = self._updatesFeedUrl
				
				elseif self._isCatalogPuller then
				
					-- Get the URL for the catalog feed.
				
					updateUrl = self._catalogSpec:catalogUpdatesUrl()
				
				else
				
					self._logger:errorf( "No update URL specified for %s", self._description )
				
				end
			
			end

			return updateUrl
			
		end )

		-- Switch over to a stream of update feed links that we gather when
		-- processing updates.
	
		:flatMap( function( updateUrl )
		
			updateLocalFeedLink( self, updateUrl )
			
			if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) then
				return Observable.never()	
			end
			
			return self._localNextUpdateFeedLink
			
		end )

		:distinctUntilChanged()

		:flatMap( function( updateUrl )
			-- Using flatMap instead of delay to avoid ordering issues in unit tests caused by using delay.
			return maybeAddDelayRequestByOz( self )
			:map( function()
				return updateUrl
			end )
		end )

		-- Get next page of update feed from Oz.

		:flatMap( function( updateUrl )
		
			self._isPullerFetchingUpdatesForSpace:onNext( true )
		
			local options = {
				bypassRateLimiter = true,
			}

			if self._documentBatchCount then
			
				-- Specify a limit parameter on the update feed. WFURI doesn't work
				-- well here because we're not working with a complete URI; we're only
				-- working with the path/query components. 
			
				if not updateUrl:find( "%?" ) then
					-- no params in the orignal url
					updateUrl = string.format( "%s?limit=%s", updateUrl, self._documentBatchCount )
				elseif not updateUrl:find( "limit=" ) then
					-- original url already has params; inject the limit as the first one
					updateUrl = updateUrl:replace( "?", 
							string.format( "?limit=%s&", self._documentBatchCount ) )
				end
				
			end
			
			if WFCloseSupport.hasBeenOrWillBeClosed( self._catalog ) or
				WFCloseSupport.hasBeenOrWillBeClosed( self._catalogSpec ) or 
				WFCloseSupport.hasBeenOrWillBeClosed( self._catalogSpec:wfOzUser() ) then
					
				-- SPAM [jandrew 2020-03-26]: If we are shutting down and the continuous task runs
				-- we can stop it from sloppily attempting additional work by returning an Observable
				-- that won't complete. In the case of https://jira.corp.adobe.com/browse/WF-939, the
				-- WFOzUser obj was being shut prior to to the call to `get` below. This was causing
				-- lots of noisy errors.
				
				return Observable.never()
				
			end


			return WFPerformanceReport.wrapObservable( 'WFCatMetadataPuller : get update feed', 
				self._catalogSpec:get( updateUrl, options ), 
				{ 
					requestUrl = updateUrl,
				} )
			
		end )

		-- Look for requested Retry-After and Log results.

		:tap( function( rawJson, headers )
			
			maybeSetRetryDelayRequestByOz( self, headers )

			self._logger:debugf( "%s: Update feed pull\n\nGET %s\n\n%s\n\n", self._description, headers, rawJson )

		end )
		
		:tapOnError( function( rawErrorJson, headers )

			local statusCode = KSTableUtils_derefOpt( headers, 'status_code' )
			
			self._logger:warnf( "%s: %s error on GET %s\n\n%s\n\n", self._description, statusCode, headers, rawErrorJson )
			
			if statusCode == '404' then
			
				local success, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, rawErrorJson )
			
				if success then
			
					local errorKey = 'catalog'
			
					if self._spaceId then
						errorKey = 'space'
					end
			
					local catalogOrSpaceErrors = KSTableUtils_derefOpt( response, 'errors', errorKey ) or {}
					
					if type( catalogOrSpaceErrors ) ~= 'table' then
						catalogOrSpaceErrors = { catalogOrSpaceErrors }
					end
				
					for _, err in ipairs( catalogOrSpaceErrors ) do
					
						if err == "does not exist" then
						
							self._isOzCatalogDeleted:onNext( true )
							break
							
						end
					
					end
				
				end

			end
			
		end )

		-- Convert to JSON.

		:map( KSJsonDecoder.decodeWebSafeJson_stripNull )

		-- Convert Oz-flavored documents to WF-flavored documents. 

		:map( function( json )
		
			local ozUpdateDocs = KSTableUtils_derefOpt( json, 'updates' )
			local baseUrl = KSTableUtils_derefOpt( json, 'base' )
			local wfUpdateDocs = convertUpdatesToDocValues( self, ozUpdateDocs, baseUrl )
			local nextLink = KSTableUtils_derefOpt( json, 'links', 'next', 'href' )
			
			if nextLink and nextLink:len() > 0 then
			
				nextLink = resolveToAbsoluteUrl( baseUrl, nextLink )
				
				wfUpdateDocs.nextFeedLink = nextLink

				if self._isPrefetchEnabled then
					-- We update the local Oz update feed link, which allows us to pre-fetch
					-- the next HTTP request while the current request continues to process.

					-- Note: We don't update the link tracked by the doc store until the http results
					-- from the previous request have been processed and the doc store has been
					-- updated. Order matters.

					if not onlyGetUpdatesOnce then

						updateLocalFeedLink( self, nextLink )

					end

				end
			
			end

			if onlyGetUpdatesOnce and numRequiredUpdates and #wfUpdateDocs < numRequiredUpdates then
				-- Return an empty updates array with no nextFeedLink so we will re-request
				-- with the same updated_since until we get the required number of updates.
				return {}
			end

			return wfUpdateDocs
			
		end )
		
		:concatMap( function( wfUpdateDocs )

			local labelUpdateObservables = {}

			for i, wfUpdateDoc in ipairs( wfUpdateDocs ) do
				if wfUpdateDoc.type == 'warning' and wfUpdateDoc.subtype == 'RevisionCompressed' then

					-- jrechtie: 2017/12/05
					-- Compressed revisions can generally be ignored.

					-- One notable exception though:
					-- a revision which has been uploaded to Oz, but not
					-- received back from Oz might get compressed away by Oz. Such
					-- a revision might be the correct three-way-merge base revision to
					-- resolve further local and remote edits but wouldn't be considered
					-- because it doesn't appear in the remote edit parent revision list.

					-- This can be fixed in docStore:setDocValue() by consulting the 'oz'
					-- label when choosing a local parentRevision for a new revision which is
					-- incoming from remote. For this to work, we need to update the
					-- 'oz' label for a compressed away revision if it's also available
					-- locally and should carry the label but doesn't because it never
					-- came back from Oz.

					local spm = wfUpdateDoc._serverProvidedMetadata
					if spm and spm.source and spm.source.id then
						local removed_revisions = KSTableUtils.slice( wfUpdateDoc.original_revisions, 2, #wfUpdateDoc.original_revisions-1, 1 )
						local labelUpdateObservable = self._catalog:docStore():updateLabelFromRevisionList( spm.source.id, 'oz', removed_revisions )
														:catch( Observable.empty() )
														:defaultIfEmpty( 'done' )
						table.insert( labelUpdateObservables, labelUpdateObservable )
						wfUpdateDoc._deleted = true
					end

				end
			end

			if #labelUpdateObservables > 0 then
				local  combinedLabelUpdates = Observable.combineLatest( unpack ( labelUpdateObservables ) )
				return Observable.combineLatest( function( wfUpdateDocs, resultFromLabelUpdates )
							return wfUpdateDocs
						end,
						Observable.just(wfUpdateDocs), combinedLabelUpdates )
			else
				return Observable.just( wfUpdateDocs )
			end
		end)

		-- Write documents to database. We use concatMap here to make sure
		-- that we're processing the results from Oz in the order that the HTTP
		-- requests were issued.

		:concatMap( function( wfUpdateDocs )

			local updateDocCount = #wfUpdateDocs
			
			local nextLink = KSTableUtils_derefOpt( wfUpdateDocs, 'nextFeedLink' ) or false
			
			local nextLinkDoc
			if nextLink then 
				nextLinkDoc = {
					_id = self._nextUpdateFeedLinkLocalDocId,
					link = stripForcedUpgrade( nextLink ),
				}
			end
		
			if updateDocCount > 0 then
				
				self._logger:debugf( "%s: About to write new docs\n\n%s\n\n", self._description, wfUpdateDocs )
				
				self._lastUpdateCycleWithData = KSDate.timeInIso8601FormatWithTimeZone()
				
				if not self._spaceId then
					self._continuousPullTask:taskDidWork()
				end
				
				self._totalDocumentsReceived = self._totalDocumentsReceived + updateDocCount
				
				local bulkSetDocValueWithNextLink = Observable.combineLatest(
					function( bulkUpdateStatus, nextFeedLink )
						return bulkUpdateStatus, nextFeedLink
					end,
					bulkSetDocValuesPreservingModificationsToDeletedItemsMovedToNewIds( self._catalog, wfUpdateDocs, { nextLinkDoc } ),
					Observable.just( nextLink ) )
			
				return WFPerformanceReport.wrapObservable( 'WFCatMetadataPuller : updateDocStore', 
					bulkSetDocValueWithNextLink, 
					{ 
						docsToUpdate = updateDocCount,
					} )
					:take( 1 )
					:tapOnCompleted( function()
						self._latestUpdatesStream:onNext( wfUpdateDocs )
					end )
			
			else
			
				markInitialSyncCompleteIfNecessary( self, docStore )
				
				local maybeSetNextLink
				if nextLinkDoc then
					-- really we only need to do this if it changed from the last next link
					maybeSetNextLink = docStore:setLocalDocValue( nextLinkDoc )
				else
					maybeSetNextLink = Observable.empty()
				end
				
				return maybeSetNextLink:concat( Observable.just( false, nextLink ) )
						-- false is just a placeholder that indicates "no documents."
						
			end

		end )
		:map( function( bulkUpdateStatus, maybeNextLink )
			
			-- Log any successes.
			self._logger:debugf( "%s: Bulk Update Status %s", self._description, bulkUpdateStatus )
			self._successfulUpdateCycles = self._successfulUpdateCycles + 1
				
			
				
			if not self._isPrefetchEnabled then
				-- We update the local Oz update feed link.
				if maybeNextLink and not onlyGetUpdatesOnce then
				
					updateLocalFeedLink( self, maybeNextLink )
			
				end
			end
			
			return bulkUpdateStatus, maybeNextLink
		end )
		:tapOnError(
			function( rawJsonOrError, headers, ... )

				-- Log any failures.
				self._logger:errorf( "%s: ERROR on update feed pull\n\nGET %s\n\n%s\n\n", self._description, headers, rawJsonOrError )

			end )

		-- If bulkUpdateStatus has content, we received updates from Oz.
		-- If we didn't receive any updates, we want to shut down  this observable
		-- and then check for updates again in a second or two.
		
		:filter( function( bulkUpdateStatus )
		
			local noMoreUpdates = not bulkUpdateStatus
			
			return noMoreUpdates 
					or onlyGetUpdatesOnce 
					or ( self._syncIsStopped and not self._spaceId )
			
		end )

		:take( 1 )

		:flatMap( function( bulkUpdateStatus )
		
			self._isPullerFetchingUpdatesForSpace:onNext( false )
			
			self._gettingLatestUpdates = nil

			if onlyGetUpdatesOnce == 'onceWithStatus' then
				return Observable.just( bulkUpdateStatus )
			end

			return Observable.empty()	
			
		end )
		
		:tapOnError( function()
		
			self._isPullerFetchingUpdatesForSpace:onNext( false )
			self._gettingLatestUpdates = nil
			
		end )
		
		return self._gettingLatestUpdates

end

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

local function maybeCreateTriggerTaskSubscription( self )

	if self._spaceId then
		return
	end

	if not self._triggerTaskSubscription then
	
		-- Every second check to see if we need to trigger the puller to check for
		-- updates based on when `checkOzForUpdates` was called.
	
		self._triggerTaskSubscription = Observable.timer( 1, 1, WFRxScheduler.sharedScheduler() )
			:subscribe( function()
					
				local currentTime = timeNow()
		
				local shouldTriggerPuller = false
		
				for checkOzAtTime, _ in pairs( self._checkOzAtTimestamps ) do
				
					if checkOzAtTime < currentTime then
						shouldTriggerPuller = true
						self._checkOzAtTimestamps[ checkOzAtTime ] = nil
					end
				
				end
				
				if shouldTriggerPuller then
					self._continuousPullTask:triggerTask()
				end
				
				if next( self._checkOzAtTimestamps  ) == nil then
				
					-- Shutdown this timer to avoid work when the timestamp table is
					-- emtpy.
					
					self._triggerTaskSubscription:dispose()
					self._triggerTaskSubscription = nil
					
				end
				
			end )
			
	end

end

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

local WFCatMetadataPuller = {

	_catalog = false,
		-- (WFCatalog)

	_description = false,
		-- (string) A unique identifier for an instance of a puller that can
		-- be used to distinguish it within logging statements.

	_catalogSpec = false,
		-- (WFCatSpec)

	_spaceId = false,
		-- (string optional) If set, it is the space id that the puller will
		-- pull updates against.

	_updatesFeedUrl = false,
		-- (string optional) If set, it is the URL for the updates feed. 

	_documentsReceivedByType = false,
		-- (table)
	
	_lastUpdateCycleWithData = false,
		-- (string) Timestamp of the last pull from Oz that included data. Used only for debugging purposes.
		
	_totalDocumentsReceived = 0,
		-- (number)
		
	_successfulUpdateCycles = 0,
		-- (number)
	
	_assetTypeHandler = false,
		-- (table)
		-- key: asset type string
		-- value: a user specified function
	
	_localNextUpdateFeedLink = false,
		-- (Rx ReplaySubject) A local stream that fires with Oz update feed links.
		
	_latestUpdatesStream = false,
		-- (Rx ReplaySubject) A stream that fires with new the latest updates
		-- that are synced from Oz.
		
	_isPullerFetchingUpdatesForSpace = false,
		-- (Rx ReplaySubject) A stream that fires with true with the puller is busy fetching updates
		-- for the space. Since we don't use the continuous pull task to fetch space-feed updates,
		-- we can't query the WFRxContinuousTask object to ask if it's doing work.
	
	_gettingLatestUpdates = false,
		-- (Rx Observable) The observable being used to pull metadata updates.
	
	_continuousPullTask = false,
		-- (WFRxContinuousTask) A continuous task to handle continuous metadata updates
	
	_isDoingInitialSync = false,
		-- (boolean) indicates whether the puller is still processing the initial sync

	_hasCheckedForInitialSync = false,
		-- (boolean) indicates whether we've checked for initial sync
	
	_hasEnsuredCatalogExists = false,
		-- (boolean) indicates whether we've already made sure catalog doc exists locally
	
	_retryDelayRequestByOz = false,
		-- (number) when set indicates seconds since _retryDelayRequestByOzAtTime to delay before requesting updates.

	_retryDelayRequestByOzAtTime = false,
		-- (number) the value of timeNowForAnimation clock when _retryDelayRequestByOz was last set.

	_syncIsStopped = false,
		-- (boolean) indicates whether metadata sync should be running, for internal use only
		
	_isOzCatalogDeleted = false,
		-- (Rx.ReplaySubject) Indicates whether the Oz catalog got deleted
		
	_isPrefetchEnabled = false,
		-- (boolean) enables the prefetching of the next HTTP batch before the current batch
		-- is completely stored.
		-- SSD based storage on mobiles is usually fast enough to keep up with network throughput,
		-- this is not necessarily true on desktop platforms with HDD based storage.

	_documentBatchCount = false,
		-- (number) Number of documents to download at once from Oz.
		
	_triggerTaskSubscription = false,
		-- (Subscription) A timer-based subscription that triggers the puller to check for new
		-- updates via WFCatMetadataPuller:checkOzForUpdates(). 
		
	_checkOzAtTimestamps = false,
		-- (table) A table of times at which we should check Oz for new updates.
	
	_nextUpdateFeedLinkLocalDocId = false,
		-- (string)
		
	_isDoingInitialSyncLocalDocId = false,
		-- (string)
		
	_isCatalogPuller = false,
		-- (boolean) True if the puller is fetching catalog-feed updates. False if it's a space-feed
		-- puller or a feed-of-feeds puller.
		
	_documentsReceivedByTypeStream = false,
		-- (Rx BehaviorSubject)
		
	_recentlyPulledDocIds = false,
		-- (WFRingBuffer)
		
	_logger = false,
		-- (WFLogger)
	
	class = {
		getNextUpdateFeedLinkLocalDocId = function( _, ... ) return getNextUpdateFeedLinkLocalDocId( ... ) end,
		getIsDoingInitialSyncLocalDocId = function( _, ... ) return getIsDoingInitialSyncLocalDocId( ... ) end,
	},
	
	WFCloseSupport.addCloseSupportMethods,

}

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

--[[!
	\brief Initialize a new WFCatMetadataPuller object.
	
	\param catalog (WFCatalog)
	
	\param description (string) An identifying string that can be used to distinguish
		a single instance of a puller from other pullers. 
		Examples: "Catalog Puller", "Feed-of-Feeds Puller", "Space <space id> Puller"
		
		!!! IMPORTANT: This description is used to identify the the local doc value
		associated with the space. Changing the description will cause the puller
		to lose track of it's local doc value it uses to track the next update
		feed link !!!
		
	\param optArgs (optional table): valid keys are as follows:
		spaceId (string)
		updatesFeedUrl (string)
		isCatalogPuller (boolean)
		catSpec (WFCatSpec)
		prefetch (boolean)
		documentBatchCount (number)
		continuousTaskDelays (table)
		
	\return WFCatMetadataPuller
--]]

function WFCatMetadataPuller:__init( catalog, description, optArgs )

	check( 'WFCatMetadataPuller', 'WFCatalog', '+string', '?table' )

	optArgs = optArgs or {}

	self:setOwned {
		_catalog = catalog:relayObject()
	}

	self._logger = WFLogger( string.format( "WFCatMetadataPuller:%s", description ) )
	
	self._description = description

	self._spaceId = optArgs.spaceId
	self._updatesFeedUrl = optArgs.updatesFeedUrl
	
	self._isCatalogPuller = optArgs.isCatalogPuller or false
	
	if self._spaceId then
		assert( not not self._updatesFeedUrl, "If a space ID is provided, so must an update URL for the space." )
	end
	
	self._nextUpdateFeedLinkLocalDocId = getNextUpdateFeedLinkLocalDocId( self._isCatalogPuller, self._description )
	self._isDoingInitialSyncLocalDocId = getIsDoingInitialSyncLocalDocId( self._isCatalogPuller, self._description )
	
	self._syncIsStopped = true
	
	self._catalogSpec = optArgs.catSpec or catalog:ozCatalogSpec()

	self._documentsReceivedByType = {}

	self._assetTypeHandler = {}

	self._localNextUpdateFeedLink = ReplaySubject( 1 )

	self._latestUpdatesStream = ReplaySubject( 1 )

	self._isPullerFetchingUpdatesForSpace = ReplaySubject( 1 )
	self._isPullerFetchingUpdatesForSpace:onNext( false )
	
	self._isOzCatalogDeleted = ReplaySubject( 1 )
	self._isOzCatalogDeleted:onNext( false )

	self._isPrefetchEnabled = optArgs.prefetch or false

	self._documentBatchCount = optArgs.documentBatchCount

	WFDiagnosticLog.addLogProvider( self )

	self._checkOzAtTimestamps = {}

	maybeCreateTriggerTaskSubscription( self )

	-- TODO [scouten 2015-07-21]: Inspect the error and network
	-- conditions. Maybe we need to wait longer?

	-- Space-feed pullers do not fetch updates continuously.
	
	if not self._spaceId then

		local continuousTaskDelays = optArgs.continuousTaskDelays or {
			progressiveBackoff = true,
			backoffStrategy = WFRxContinuousTask.BackoffStrategy_Exponential,
			noWorkDelay = 2,
			maxNoWorkDelay = 30,
			errorDelay = 10,
			maxDelay = 60,
		}
		
		local continuousTask = function() return getLatestUpdates( self ) end

		self._continuousPullTask = WFRxContinuousTask( "WFCatMetadataPuller_"..catalog:ozCatalogId(), continuousTask, continuousTaskDelays )
		
	end
	
end

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

function WFCatMetadataPuller:close()

	check( 'WFCatMetadataPuller' )
	
	local continuousPullTask = self._continuousPullTask
	self._continuousPullTask = nil
	
	if continuousPullTask then
		continuousPullTask:close()
	end

	local localNextUpdateFeedLink = self._localNextUpdateFeedLink
	self._localNextUpdateFeedLink = nil
	
	if localNextUpdateFeedLink then
		localNextUpdateFeedLink:dispose()
	end

	local latestUpdatesStream = self._latestUpdatesStream
	self._latestUpdatesStream = nil
	
	if latestUpdatesStream then
		latestUpdatesStream:dispose()
	end
	
	local isOzCatalogDeleted = self._isOzCatalogDeleted
	self._isOzCatalogDeleted = nil
	
	if isOzCatalogDeleted then
		isOzCatalogDeleted:dispose()
	end

	local triggerTaskSubscription = self._triggerTaskSubscription
	self._triggerTaskSubscription = nil
	
	if triggerTaskSubscription then
		triggerTaskSubscription:dispose()
	end
	
	local isPullerFetchingUpdatesForSpace = self._isPullerFetchingUpdatesForSpace
	self._isPullerFetchingUpdatesForSpace = nil

	if isPullerFetchingUpdatesForSpace then
		isPullerFetchingUpdatesForSpace:dispose()
	end
	
	local documentsReceivedByTypeStream = self._documentsReceivedByTypeStream
	self._documentsReceivedByTypeStream = nil
	
	if documentsReceivedByTypeStream then
		documentsReceivedByTypeStream:dispose()
	end
	
end

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

-- SPAM [scouten 2015-07-21]: This is really only exposed for testing purposes.
-- Hide later?

function WFCatMetadataPuller:getUpdatesOnce()

	check( 'WFCatMetadataPuller' )

	return getLatestUpdates( self, true )

end

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

function WFCatMetadataPuller:getLatestUpdates( onceRequest, numRequiredUpdates )

	check( 'WFCatMetadataPuller' )

	return getLatestUpdates( self, onceRequest, numRequiredUpdates )

end

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

--[[!
	\brief Get the latest updates, but only if the current update link is newer
		than the next update link that we track locally.
		
		NOTE: This API is only called by the feed-of-feeds code in WFCatSpaceFeedsManager.
		
	\param currentUpdateLink (optional string) Oz link that would be received as next link
		if sync were completely up to date, i.e. response would be an empty list. 

	\return void
--]]


function WFCatMetadataPuller:getLatestUpdatesIfNeeded( currentUpdateLink )

	check( 'WFCatMetadataPuller', '?string' )

	-- currentUpdateLink is the latest update link as indicated within the space
	-- document. If our nextUpdateFeedLink does not match currentUpdateLink,
	-- then it means we are behind Oz and should fetch the latest updates.

	local result = Observable.just()

	if currentUpdateLink then
	
		result = self._catalog:docStore():localDocValue( self._nextUpdateFeedLinkLocalDocId )
	
			:take( 1 )
	
			:flatMap( function( docValue )
			
				local maybeUpdatesUrl = KSTableUtils_derefOpt( docValue, 'link' )
				
				if maybeUpdatesUrl ~= currentUpdateLink then
				
					return getLatestUpdates( self )
				
				else
				
					return Observable.just()
				
				end
				
			end )
	
	else

		result = getLatestUpdates( self )

	end

	return result

end

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

function WFCatMetadataPuller:stopContinuousUpdates()

	check( 'WFCatMetadataPuller' )

	self._syncIsStopped = true
	
	if not self._spaceId then
		self._continuousPullTask:stopContinuousTask()
	end
	
end

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

function WFCatMetadataPuller:getUpdatesContinuously()

	check( 'WFCatMetadataPuller' )

	self._syncIsStopped = false
	
	if not self._spaceId then
		self._continuousPullTask:startContinuousTask()
	end
	
end

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

function WFCatMetadataPuller:isDoingWork()

	check( 'WFCatMetadataPuller' )

	if self._spaceId then
		return self._isPullerFetchingUpdatesForSpace
	else
		return self._continuousPullTask:isTaskDoingWork()
	end
	
end

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

function WFCatMetadataPuller:onceInitialSyncIsComplete()

	check( 'WFCatMetadataPuller' )

	return self._catalog:docStore()
		
		:ongoingLocalDocValue( self._isDoingInitialSyncLocalDocId )
		
		:filter( function( docValue )
			
			-- isDoingInitialSync is nil or true if initial sync hasn't started or
			-- is in progress.
			
			return docValue.isDoingInitialSync == false
			
		end )
		
		:take( 1 )
	
end

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

--[[!
	\brief Force the puller to check Oz for updates. We don't do this right away,
		but rather we wait 5 seconds to check Oz for updates due to Oz's built-in
		5 second delay. This should be called when we expect there to be new updates,
		such as after we push metadata changes.

	\return void
--]]

function WFCatMetadataPuller:checkOzForUpdates()

	check( 'WFCatMetadataPuller' )

	local OZ_DELAY_IN_SECONDS = 5
	
	local OZ_DELAY_BUFFER = 2
		-- Allow for a short delay to make it  more likely that Oz will have a response
		-- for us.

	local checkOzAtTime = timeNow() + OZ_DELAY_IN_SECONDS + OZ_DELAY_BUFFER

	checkOzAtTime = math.floor( checkOzAtTime + 0.5 )
		-- This effectively limits these updates to one per second.
		
	self._checkOzAtTimestamps[ checkOzAtTime ] = true 
		
	maybeCreateTriggerTaskSubscription( self )
		
end

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

function WFCatMetadataPuller:isOzCatalogDeleted()

	check( 'WFCatMetadataPuller' )
	
	return self._isOzCatalogDeleted
		:distinctUntilChanged()
		
end

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

--[[!

	\brief Add a custom asset type handler to the WFCatMetadataPuller

	\param assetType (string) The oz asset type the custom handler will act upon.
	
	\param handlerFn( (function) A function that will be called on each asset
	of the given assetType.
	
	handlerFn takes two arguments:
		1. A WFCatMetadataPuller instance.
		2. The constructed doc value that has yet to be inserted into the doc store.
	
	Example handlerFn:
	
		local function addOpinionOnPieAnnotation( wfCatMetadataPuller, docValue )

			docValue._localOnly = { 
				opinionOnPie = "Pretty good."
			}

		end
	
	\return void
	
--]]

function WFCatMetadataPuller:addAssetTypeHandler( assetType, handlerFn )

	check( 'WFCatMetadataPuller', 'string', 'function' )

	if self._assetTypeHandler[ assetType ] then
	
		self._logger:warnf( "%s: Now overwriting the existing asset type handler", self._description )
	
	end
	
	self._assetTypeHandler[ assetType ] = handlerFn

end

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

function WFCatMetadataPuller:getLatestDownloadedAssetIdsStream()

	return self._latestUpdatesStream
	
		:map( function( updates )
			
			local assetIds = {}
			
			for _, updateDoc in ipairs( updates ) do
			
				if isValidAssetDocument( updateDoc ) then
					table.insert( assetIds, updateDoc._id )
				end
			
			end
			
			return assetIds
			
		end )
		
end

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

local function createDiagnosticSummaryStr( pullerDescription, documentsReceivedByType )

	local numberOfUpdatesSynced 
			= KSTableUtils_derefOpt( documentsReceivedByType, 'update', 'total' ) or 0
	local numberOfDeletionsSynced  
			= KSTableUtils_derefOpt( documentsReceivedByType, 'delete', 'total' ) or 0

	return string.format( "%s synced down %d updates, %d deletions from Oz.", 
			pullerDescription, numberOfUpdatesSynced, numberOfDeletionsSynced )
			
end

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

local function justReturn( ... )
	return ...
end

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

--[[!
	\brief This API that can be used gather real-time diagnostic data from the
		metadata puller. 
		
		Note: At this time, it's not possible to estimate progress for the metadata 
		puller. There's currently no way to determine the number of outstanding 
		updates that we need to pull from Oz. See the following for more info:
		https://lightroom.slack.com/archives/GM8LXNGFL/p1616097899003900

	\param optCalm (optional number) The maximum frequency at which the stream will
		fire with updates. Defaults to 1 second.

	\param optBufferSize (optional number) The size of the ring buffer used to store
		recently synced document Ids. Defaults to 100.
	
	\return Observable< table > An Observable whose value tracks a table of
		sync diagnostic data. An example data event may look like this:
		
		{
			documentsSyncedByType = {
				delete = {
					asset = {
						NULL = 1,
					},
					total = 1,
				},
				update = {
					album_asset = {
						NULL = 2,
					},
					asset = {
						deleted_image = 1,
						image = 5,
					},
					total = 8,
				},
			},
			isCurrentlySyncing = true,
			recentlySyncedDocIds = {
				"e1f62d81001d452bb33cce3e46f906e5",
				"149d3f4af51045eea912590e399f920f",
				"983605b2a9934e328df455a4d8a2278f",
				"1387f278615545c3aac47a1a1a6dd0a5",
				"149d3f4af51045eea912590e399f920f",
				"f93f5da48e2848018ba9c64a9f0380c2",
				"52b8d7c5dedf4e2b8c2d9da1d29f063d",
				"ba88ded724814dfaa94e35bb410b2b25",
				"6dce02ee42b1493b84f5665ae4baa06c",
			},
			summaryStr = "Catalog Puller synced down 8 updates, 1 deletions from Oz.",
			totalDocumentsSynced = 9,
		}
--]]

function WFCatMetadataPuller:syncDiagnostics( optCalm, optBufferSize )

	check( 'WFCatMetadataPuller', '?number', '?number' )

	if not self._documentsReceivedByTypeStream then
		self._documentsReceivedByTypeStream = BehaviorSubject( {} )
	end
	
	if not self._recentlyPulledDocIds then
		self._recentlyPulledDocIds = WFRingBuffer.make( optBufferSize or 100 )
	end
	
	return Observable.combineLatest(
			self._documentsReceivedByTypeStream,
			self:isDoingWork() )

		:map( function( documentsReceivedByType, isCurrentlySyncing )
		
			local diagnostics = {
				summaryStr = createDiagnosticSummaryStr( self._description, documentsReceivedByType ),
				totalDocumentsSynced = self._totalDocumentsReceived,
				isCurrentlySyncing = not not isCurrentlySyncing,
				documentsSyncedByType = documentsReceivedByType,
				recentlySyncedDocIds = WFRingBuffer.map( justReturn, self._recentlyPulledDocIds )
			}
			
			self._logger:debugf( "%s syncDiagnostics update: %s", self._description, diagnostics )
			
			return diagnostics
				
		end )

end

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

function WFCatMetadataPuller:getDiagnosticLogData()

	return {
		isCurrentlySyncing = not self._syncIsStopped,
		documentsReceivedByType = self._documentsReceivedByType,
		successfulUpdateCycles = self._successfulUpdateCycles,
		nextUpdateFeedLink = self._catalog:docStore():localDocValue( self._nextUpdateFeedLinkLocalDocId ),
		totalDocumentsReceived = self._totalDocumentsReceived,
		lastUpdateCycleWithData = self._lastUpdateCycleWithData,
		isOzCatalogDeleted = self._isOzCatalogDeleted,
		recentlySyncedDocIds = self._recentlyPulledDocIds and WFRingBuffer.map( justReturn, self._recentlyPulledDocIds ) or nil,
	}

end

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

function WFCatMetadataPuller:sendDiagnosticLogData()

	return {

		name = 'Metadata Sync Down: ' .. self._catalogSpec:catalogId(),
		id = 'WFCatMetadataPuller-' .. self._catalogSpec:catalogId(),
		weight = 500,
		parentSection = WFDiagnosticLog.Sections.SYNC,
		
		body = self:getDiagnosticLogData()

	}

end

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

-- SPAM [jandrew 2019-08-29]: This function is only used by unit tests.

function WFCatMetadataPuller:_testOnly_getIsTaskRunningStream()

	check( 'WFCatMetadataPuller' )

	if self._spaceId then
		return Observable.just( false )
	end

	return self._continuousPullTask:_testOnly_getIsTaskRunningStream()

end

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

return Rx.Class 'WFCatMetadataPuller'( WFCatMetadataPuller )
