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

WFCatMetadataPusher.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.logger
local WFLogger = require 'WFLogger'

	-- 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 WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFConfiguration = require 'WFConfiguration'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRingBuffer = require 'WFRingBuffer'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxRetryObservable = require 'WFRxRetryObservable'
local WFRxScheduler = require 'WFRxScheduler'
local WFTableMerge = require 'WFTableMerge'

	-- WF.oz_client.managed_catalog
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatSyncStatus = require 'WFCatSyncStatus'
local WFCatTrashUtils = require 'WFCatTrashUtils'

	-- WF.oz_client.lrmobile
local OzSyncMetadataMapper

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

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

local check = assert( WFArguments.check )

local derefOpt = KSTableUtils.derefOpt
local runUntilDone = WFRxRunUntilDone
local timeNowForAnimation = assert( KSDate.timeNowForAnimation )

local logger = WFLogger 'WFCatMetadataPusher'
--:setLevel( 'info' )

if not WFPlatformInfo.isDesktop then
	OzSyncMetadataMapper = require 'OzSyncMetadataMapper'
end

local isTrashAssetDocument = assert( WFCatIndexUtils.isTrashAssetDocument )
local WFCatTrashUtils_createFromInfo = assert( WFCatTrashUtils.createFromInfo )
local WFCatTrashUtils_isDeleteTrash = assert( WFCatTrashUtils.isDeleteTrash )
local WFCatTrashUtils_isDirectlyDelete = assert( WFCatTrashUtils.isDirectlyDelete )
local WFCatTrashUtils_getCreateWithRevisionId = assert( WFCatTrashUtils.getCreateWithRevisionId )

local kNotEnoughUpdatesError = "Error not enough updates"

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

local function getContinueFromLocalDocValueKey( _, isSpacePusher )

	local key = 'Oz-sendUpdatesContinueFrom'

	if isSpacePusher then
	
		key = 'Oz-spaceSendUpdatesContinueFrom'
	
	end
	
	return key

end

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

local updateSendUpdatesContinueFrom = function( self )
	
	return self._catalog:docStore():localDocValue( self._continueFromLocalDocValueKey )

		:flatMap( function( docValue )
			if self._isSpacePusher and not KSTableUtils.derefOpt( docValue, 'continueFrom' ) then
				return self._catalog:docStore():localDocValue( getContinueFromLocalDocValueKey( nil, false ) )
			else
				return Observable.just( docValue )
			end
		end )
		:map( function( docValue )
			local continueFrom = KSTableUtils.derefOpt( docValue, 'continueFrom' ) or 0

			-- If pusher is closed before we get a localDocValue
			-- _sendUpdatesContinueFrom can be nil-ed out.
			if self._sendUpdatesContinueFrom then
				self._sendUpdatesContinueFrom:onNext( continueFrom )
			end

			logger:infof( "Oz-sendUpdatesContinueFrom -> continueFrom = %s", continueFrom )

			return continueFrom

		end )

end

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

local function combineChangedDocsWithOz( docStore, changedDocs )

	local docIds = { label = 'oz' }

	for i, docValue in ipairs( changedDocs ) do
		docIds[ i ] = docValue._localId
	end

	return docStore:bulkDocValue( docIds )
		:map( function( docValues )
			return changedDocs, docValues
		end )

end

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

-- The following document types will not cause logger warnings if they have local
-- changes and are not in the allowlist of accepted document types for a particular
-- metadata pusher instance.

local doNotWarnDocTypes = {
	feedbackuser = true,
}

-- The following document types are supported via the catalog bulk update API on Oz

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

local createDocTypeAllowlist = {
	asset = true,
	album = true,
	album_asset = true,
	space = true,
	space_album = true,
	comment = true,
	favorite = true,
	stack_asset = true,
	spaceinvite = true,
	version = true,
	watermark = true,
}

local updateDocTypeAllowlist = {
	catalog = true,
	asset = true,
	album = true,
	album_asset = true,
	space = true,
	stack_asset = true,
	spaceinvite = true,
	space_user = true,
	version = true,
	watermark = true,
}

local deleteDocTypeAllowlist = {
	asset = true,
	album = true,
	album_asset = true,
	space = true,
	space_album = true,
	spaceinvite = true,
	space_user = true,
	comment = true,
	favorite = true,
	stack_asset = true,
	version = true,
	watermark = true,
}

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

-- The following document types are supported via the space-feed bulk update API on Oz

local spaceFeedCreateDocTypeAllowlist = {
	asset = true,
	album_asset = true,
	comment = true,
	favorite = true,
	stack_asset = true,
	version = true,
}

local spaceFeedUpdateDocTypeAllowlist = {
	space_user = true,
	asset = true,
	version = true,
}

local spaceFeedDeleteDocTypeAllowlist = {
	asset = true,
	album_asset = true,
	comment = true,
	favorite = true,
	stack_asset = true,
	version = true,
}

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

local function docUsesSpaceInCloud( doc )

	-- Note that trash assets do not use space in the cloud they will have
	-- subtype 'deleted_image' and 'deleted_video'.
	return not not (  doc and doc.type == 'asset' and
		( doc.subtype == 'image' or doc.subtype == 'video' ) )

end

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

local album_assetCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.asset = localDoc.asset
	docToSend.album = localDoc.album
	
	payload.asset = nil
	payload.album = nil
	
end

local space_albumCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.space = localDoc.space
	docToSend.album = localDoc.album
	
	payload.space = nil
	payload.album = nil
	
end

local stack_assetCreateRemapFn = function( self, localDoc, payload, docToSend )
	
	docToSend.asset = localDoc.asset
	docToSend.stack = localDoc.stack
	
	payload.asset = nil
	payload.stack = nil
	
end

local commentCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.space = localDoc.space
	docToSend.asset = localDoc.asset
	
	payload.space = nil
	payload.asset = nil
	
end

local favoriteCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.space = localDoc.space
	docToSend.asset = localDoc.asset
	
	payload.space = nil
	payload.asset = nil
	
end

local assetCreateRemapFn = function( self, localDoc, payload, docToSend )

	if docToSend.subtype == 'preset' or docToSend.subtype == 'profile'
								or docToSend.subtype == 'camera_profile'then
		-- for presets/profiles we don't want to let duplicates in to the system
		docToSend.if_none_match = derefOpt( payload, 'importSource', 'sha256' )
	end

	local extraTrashDelta = nil

	local createFromInfo = WFCatTrashUtils_createFromInfo( localDoc )

	if createFromInfo and createFromInfo.id == localDoc._id then
		-- Can't create a docId from itself, clear createFromInfo to issue a regular create.
		createFromInfo = nil
		logger:errorf( "ERROR createFromInfo.id == localDoc._id for localDoc=\n\n%s\n\n", localDoc )
	end

	if createFromInfo then

		if isTrashAssetDocument( localDoc ) then

			-- Create trash asset by deleting.

			-- If we are creating a trash asset and the original asset existed on
			-- oz create the trash asset by sending a delete with a deleted_asset
			-- entry instead. We do this here instead of on every tombstone to
			-- better handle repeated delete restore cases.

			local createTrashFromId = createFromInfo.id
			local createTrashFromRevisionId = createFromInfo.revision_id

			if createFromInfo and localDoc.subtype == createFromInfo.subtype then

				-- Oz only allows you to create trash from a regular asset so
				-- send the restore for this even though it has been deleted
				-- locally, it will be deleted immediately by the delta below it.

				createTrashFromId = createFromInfo.lastAssetId
				createTrashFromRevisionId = createFromInfo.lastRevisionId
				local restoredSubtype = string.match( localDoc.subtype, '^deleted_(.*)' )

				extraTrashDelta = {
					id = createTrashFromId,
					revision_id = createTrashFromRevisionId,
					type = localDoc.type,
					subtype = restoredSubtype,
					action = 'create',
					payload = {
						userUpdated = localDoc.userCreated
					},
					restore_from = {
						id = createFromInfo.id
					},
				}

			end

			-- Trash assets that originally existed on oz may have had a
			-- modification applied to them either before they were trashed
			-- or if they were restored and trashed again while offline
			-- so use the revision the trash asset was created with in case
			-- another revision was applied to it.
			local createWithRevisionId = WFCatTrashUtils_getCreateWithRevisionId( localDoc )

			-- Trash assets are created via a delete if the asset they are
			-- being created for existed in the cloud. If the original asset
			-- is not in the cloud they will be created via the create request below.
			docToSend = {
				id = createTrashFromId,
				revision_id = createTrashFromRevisionId,
				type = localDoc.type,
				action = 'delete',
				deleted_asset = {
					id = localDoc._id,
					revision_id = createWithRevisionId,
					payload = {
						userCreated = localDoc.userCreated,
						purgeDate = localDoc.purgeDate,
					}
				},
			}

		else

			-- Create regular asset by restoring.

			local createFromId = createFromInfo.id
			local createFromRevisionId = createFromInfo.revision_id

			if localDoc.subtype == createFromInfo.subtype then

				-- Oz only allows you to restore from a trash asset so
				-- send the delete for the target to create a trash asset
				-- that we can restore from.

				extraTrashDelta = {
					id = createFromId,
					revision_id = createFromRevisionId,
					type = localDoc.type,
					action = 'delete',
					deleted_asset = {
						id = createFromInfo.lastAssetId,
						revision_id = createFromInfo.lastRevisionId,
						payload = {
							userCreated = createFromInfo.lastTrashUserCreated,
							purgeDate = createFromInfo.lastTrashPurgeDate
						}
					},
				}

				createFromId = createFromInfo.lastAssetId
				createFromRevisionId = createFromInfo.lastRevisionId

			end

			-- Always restore from the initial revId in case modifications
			-- have been made to the asset locally, this is done in order
			-- to have the conflict resolver see newer revisions and cause
			-- us to send a subsequent update with the modifications.
			docToSend.revision_id = WFCatTrashUtils_getCreateWithRevisionId( localDoc )

			docToSend.restore_from = {
				id = createFromId,
			}

			docToSend.payload = {
				userUpdated = localDoc.userUpdated,
			}

		end

	else

		-- Normal create path, add original if it exists, which should
		-- only occur when a trash asset is created directly on oz
		-- because the asset was imported and deleted before being synced.

		docToSend.original = localDoc.original

	end

	-- Handle freemium asset payload on mobile
	if not WFPlatformInfo.isDesktop then
		local userId = self._catalog:ozUserId()
		
		OzSyncMetadataMapper.removeEmptyTables( payload )
		OzSyncMetadataMapper.remapGlobalDocumentId( userId, payload )
	end
	
	
	-- Return docToSend, extraTrashDelta as docToSend may be changed ot use new
	-- table.  extraTrashDelta will be nil if an extra delta isn't needed

	return docToSend, extraTrashDelta

end
	
local spaceinviteCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.space = localDoc.space
	
	payload.space = nil
	
end

local versionCreateRemapFn = function( self, localDoc, payload, docToSend )

	docToSend.asset = localDoc.asset
	payload.asset = nil
	
	docToSend.importedOnDevice = localDoc.importedOnDevice
	payload.importedOnDevice = nil
	
end

local album_assetUpdateRemapFn = function( self, localDoc, payload, docToSend )

	if derefOpt( docToSend, 'payload', 'asset' ) then
		KSTableUtils.setNested( docToSend, nil, 'payload', 'asset' )
	end
	if derefOpt( docToSend, 'payload', 'album' ) then
		KSTableUtils.setNested( docToSend, nil, 'payload', 'album' )
	end
	
end

local versionUpdateRemapFn = function( self, localDoc, payload, docToSend )

	if derefOpt( docToSend, 'payload', 'asset' ) then
		KSTableUtils.setNested( docToSend, nil, 'payload', 'asset' )
	end
	
	if derefOpt( docToSend, 'payload', 'importedOnDevice' ) then
		KSTableUtils.setNested( docToSend, nil, 'payload', 'importedOnDevice' )
	end
	
end

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

local function replaceDeleteWithJsonNull( t )

	local didReplaceAnything = false

	for k, v in pairs( t ) do

		if type( v ) == 'table' then

			if v._delete == true then
				didReplaceAnything = true
				t[ k ] = KSJsonEncoder.json_null
			else
				didReplaceAnything = replaceDeleteWithJsonNull( v ) or didReplaceAnything
			end

		end

	end

	return didReplaceAnything

end


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

local function deltaForCreateDoc( self, localDoc, newDocsFilteredDueToOzError )

	-- If document is deleted locally and no corresponding Oz doc,
	-- no need to send anything.

	if localDoc._deleted or not localDoc.type then
		return
	end

	if not self._createDocTypeAllowlist[ localDoc.type ] then
		logger:warnf( "Unable to upload %s document ID %s because not supported in bulk API",
					  localDoc.type, localDoc._id )
		return
	end
	
	local spaceId = localDoc._spaceId
	
	if ( self._userIsOverQuota or ( spaceId and self._spacesOverQuotaById[ spaceId ] ) ) 
			and docUsesSpaceInCloud( localDoc ) then
			
		-- We're out of storage on Oz and this new doc will use up space so we will pre-filter
		-- out the create and, instead of sending it to Oz, mark it with a StorageExceededError
		-- When storage is once again available on Oz, it should be cleared by the normal process.
		logger:debugf( "Filtering out create of %s due to pre-existing StorageExceededErrors", localDoc._id )

		local storageErrors = newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kStorageExceededError ] or {}
		storageErrors[ #storageErrors + 1 ] = localDoc._id
		newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kStorageExceededError ] = storageErrors

		return
		
	end

	if ( self._userIsExpired and docUsesSpaceInCloud( localDoc ) ) 
			or ( spaceId and self._spacesExpiredById[ spaceId ] ) then
			
		logger:debugf( "Filtering out create of %s due to trial or subscription expired", localDoc._id )

		local expiredErrors = newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] or {}
		expiredErrors[ #expiredErrors + 1 ] = localDoc._id
		newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] = expiredErrors

		return
		
	end

	local payload = {}

	for k, v in pairs( localDoc ) do
		if type( k ) == 'string' and not k:match( '^_' ) then
			payload[ k ] = v
		end
	end

	payload.original = nil
	payload.type = nil
	payload.subtype = nil

	local docToSend = {
		id = localDoc._id,
		revision_id = localDoc._revId,
		type = localDoc.type,
		subtype = localDoc.subtype,
		action = 'create',
		payload = payload,
	}

	local extraTrashDelta = nil

	local createRemapFn = self._createDocRemapFunctions[ localDoc.type ]

	if createRemapFn then
		local newDocToSend
		newDocToSend, extraTrashDelta = createRemapFn( self, localDoc, payload, docToSend )
		docToSend = newDocToSend or docToSend
	end

	return docToSend, extraTrashDelta

end

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

local function deltaForDeleteDoc( self, localDoc, ozDoc, newDocsFilteredDueToOzError )

	-- If it's space asset and owner account expires
	
	local spaceId = localDoc._spaceId
	
	if ( spaceId and self._spacesExpiredById[ spaceId ] ) then
			
		logger:debugf( "Filtering out delete of %s due to trial or subscription expired", localDoc._id )

		local expiredErrors = newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] or {}
		expiredErrors[ #expiredErrors + 1 ] = localDoc._id
		newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] = expiredErrors

		return
		
	end

	local extraTrashDelta = nil
	local subtype = localDoc.subtype
	
	local createFromInfo = WFCatTrashUtils_createFromInfo( localDoc )
	if subtype == 'deleted_image' or subtype == 'deleted_video' then
	
		if createFromInfo and WFCatTrashUtils_isDeleteTrash( localDoc ) then

			if createFromInfo.subtype ~= subtype then

				-- If we are deleting a trash asset that doesn't exist on oz but
				-- the original asset it referred to does exist on oz then
				-- send the delete to the original asset and follow up with a
				-- delete for the trash asset as well.

				extraTrashDelta = {
					id = createFromInfo.id,
					revision_id = createFromInfo.revision_id,
					type = localDoc.type,
					action = 'delete',
					deleted_asset = {
						id = localDoc._id,
						revision_id = localDoc._revId,
						payload = {
							userCreated = localDoc.userCreated, -- date doesn't matter we are deleting this immediately after it is created.
							purgeDate = localDoc.purgeDate,
						}
					},
				}
				
			elseif not ozDoc then

				-- We are deleting a trash asset that was restored, deleted, and
				-- then permanently deleted since the last sync. Just delete the
				-- trash asset we were going to create the new trash asset from.
				return {
					id = createFromInfo.id,
					revision_id = createFromInfo.revision_id,
					type = localDoc.type,
					action = 'delete',
				}

			end

		end

	elseif WFCatTrashUtils_isDirectlyDelete( localDoc ) and ( createFromInfo or ozDoc ) then
	
		-- This process only works for normal assets. The asset can be original asset or restored asset.
		-- Deletion would only be sent if the asset exists on Oz
	
		local assetId = createFromInfo and createFromInfo.id or localDoc._id
		local revisionId = createFromInfo and createFromInfo.revision_id or localDoc._revId
		local trashId = KSIdGenerator.uuidInLowercaseHex()
		local trashRevId = KSIdGenerator.uuidInLowercaseHex()
		local trashPurgeDate = KSDate.timeInIso8601Format()
		
		return 
			{
				id = trashId,
				revision_id = trashRevId,
				type = localDoc.type,
				action = 'delete',
			},
			{
				id = assetId,
				revision_id = revisionId,
				type = localDoc.type,
				action = 'delete',
				deleted_asset = {
					id = trashId,
					revision_id = trashRevId,
					payload = {
						userCreated = trashPurgeDate, -- date doesn't matter we are deleting this immediately after it is created.
						purgeDate = trashPurgeDate,
					}
				}
			}
	
	end

	-- If extraTrashDelta is set that means even though this asset doesn't
	-- exist on oz we need to also issue the delete for it as the
	-- extraTrashDelta will be creating it.

	if not ozDoc and not extraTrashDelta then
		-- If document is deleted locally and no corresponding Oz doc,
		-- no need to send anything.
		return
	end

	-- Explicitly calling out this case:

	-- if we sent a deletion to Oz,
	-- the returned doc won't include the revision_ids array. Hence the WFRxDocStore
	-- creates a new local revId which triggers this sync code again
	-- Adding this comparision here, makes this case clear how it is handled.
	-- Without it we only returned early as the localDoc no longer carries the type
	-- property which is catched by the deleteDocType allowlisting

	if ozDoc and ozDoc._deleted then
		logger:infof( "Both local and Oz docs deleted, not sending change", localDoc._id )
		return
	end

	local docType = localDoc.type or ( ozDoc and ozDoc.type )
	if self._deleteDocTypeAllowlist[ docType ] then

		return {
			id = localDoc._id,
			revision_id = localDoc._revId,
			type = docType,
			action = 'delete',
		}, extraTrashDelta

	else
		logger:infof( "Unable to upload deletion of %s due to non-allowlisted docType %s", localDoc._id, docType )
		return
	end

end

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

local function deltaForUpdateDoc( self, localDoc, ozDoc, newDocsFilteredDueToOzError )

	-- If it's space asset and owner account expires
	
	local spaceId = localDoc._spaceId
	
	if ( spaceId and self._spacesExpiredById[ spaceId ] ) then
			
		logger:debugf( "Filtering out update of %s due to trial or subscription expired", localDoc._id )

		local expiredErrors = newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] or {}
		expiredErrors[ #expiredErrors + 1 ] = localDoc._id
		newDocsFilteredDueToOzError[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] = expiredErrors

		return
		
	end

	-- Local document has a new revision, calculate diffs against Oz.

	local delta = WFTableMerge.createDiffForOz( ozDoc, localDoc )

	for k in pairs( delta ) do
		if type( k ) == 'string' and k:match( '^_' ) then
			delta[ k ] = nil
		end
	end

	local canUpdateDoc = self._updateDocTypeAllowlist[ localDoc.type ]
	
	if type( canUpdateDoc ) == 'function' then
		canUpdateDoc = canUpdateDoc( localDoc )
	end
	
	if not canUpdateDoc then
	
		local loggerFn = logger.warnf
	
		if doNotWarnDocTypes[ localDoc.type ] then
			loggerFn = logger.infof
		end
	
		loggerFn( logger, "Unable to update %s document ID %s because not supported in bulk API",
					  localDoc.type, localDoc._id )
		return
		
	end

	if not ozDoc._serverProvidedMetadata then
		return
	end
	local ozRevIds = derefOpt( ozDoc, '_serverProvidedMetadata', 'revision_ids' )
	
	local ozParentRevId = ozRevIds and ozRevIds[ #ozRevIds ]

	local shouldOzGenerateRenditions
	local developSection = derefOpt( delta, 'develop' )
	local xmpCameraRawSha256 = derefOpt( delta, 'develop', 'xmpCameraRaw', 'sha256' )
	if developSection and not xmpCameraRawSha256 then
		-- If there are develop changes and they're not external, then we should ask Oz to 
		-- generate the renditions. If they are external, we will ask when we upload the
		-- external settings
		shouldOzGenerateRenditions = 'all'
	end

	-- We can never update type or subtype. Drop any such changes on the floor.

	delta.original = nil
	delta.subtype = nil
	delta.type = nil
	delta.purgeDate = nil
	
	local docType = ozDoc.type or localDoc.type
	
	replaceDeleteWithJsonNull( delta )

	local docToSend = {
		id = localDoc._id,
		revision_id = localDoc._revId,
		action = 'update',
		type = docType,
		generate_renditions = shouldOzGenerateRenditions,
		if_match = ozParentRevId or nil,
		payload = delta or {},
	}

	local updateDocRemapFn = self._updateDocRemapFunctions[ localDoc.type ]
	
	if updateDocRemapFn then
		updateDocRemapFn( self, localDoc, delta, docToSend )
	end
	
	if not next( docToSend.payload ) then
	
		-- SPAM [jandrew 2021-09-23]: The following addresses bug WF-1757.
		-- Oz will reject document updates that have an empty payload.
		-- However, we've seen cases where a user has a local document
		-- revision that matches the latest revision on Oz. When we calculate
		-- the delta in this state, it's empty. If this state is detected,
		-- then we mark the revision with a label indicating we do not need
		-- to sync it.

		local revisionIsOnOz = ozDoc and ozDoc._revId == ozParentRevId
		
		if revisionIsOnOz then
		
			-- If we're certain our local revision is identical to the one on Oz,
			-- then update our local revision and add the '_dontsendtooz' label. We create
			-- a new revision here, as opposed to adding the label to the current revision,
			-- in order to increment the doc store change counter and trigger any change
			-- counter listeners.

			self._catalog:docStore():applyOverlay( {
					_id = localDoc._id,
					_newRevId = KSIdGenerator.uuidInLowercaseHex(),
					_newLabel = '_dontsendtooz'
				} )
				:call( runUntilDone )

			return

		else
		
			-- If there's nothing in the delta, and no Oz doc to analyze, we can't
			-- send it to Oz.
			
			return
		
		end

	end
	
	return docToSend
	
end

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

local function trackDocumentForDiagnostics( self, typeOfUpdate, doc )

	local docType = doc.type

	if docType then

		local subtype = doc.subtype or 'NULL'

		local function updateDocumentCount( typeOfUpdate )
		
			local total = derefOpt( self._documentsSentByType, typeOfUpdate, 'total' ) or 0

			KSTableUtils.setNested( self._documentsSentByType, total + 1, typeOfUpdate, 'total' )
		
			local docCount = derefOpt( self._documentsSentByType, typeOfUpdate, docType, subtype ) or 0

			KSTableUtils.setNested( self._documentsSentByType, docCount + 1, typeOfUpdate, docType, subtype )
		
			return self._documentsSentByType
		
		end
			
		self._documentsSentByType = updateDocumentCount( typeOfUpdate )
			
		if self._recentlyPushedDocIds then
		
			WFRingBuffer.addItem( self._recentlyPushedDocIds, doc._id )
			
		end
			
		if self._documentsSentByTypeStream then
			self._documentsSentByTypeStream:onNext( self._documentsSentByType )
		end
		
	end

end

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

local function convertOneDocToDelta( self, localDoc, ozDoc, newDocsFilteredDueToOzError )

	-- If document is in conflict, don't send. Wait for conflict resolution
	-- and we'll sort it out later.

	if localDoc._conflict then
		logger:infof( "Unable to upload %s document ID %s at changeCounter %s due to conflict",
					  localDoc.type, localDoc._id, localDoc._changeCounter )
		return
	end

	-- Is local document deleted?

	if localDoc._deleted then
		trackDocumentForDiagnostics( self, 'delete', localDoc )
		return deltaForDeleteDoc( self, localDoc, ozDoc, newDocsFilteredDueToOzError )
	end
	
	local shouldSkip, reason = WFCatOzErrors.shouldSkipUploadDueToError( localDoc )
	if shouldSkip then
		logger:infof( "Unable to upload %s document with id %s due to %s",
						localDoc.type, localDoc._id, reason )
		if localDoc.type then
			self._docsFilteredDueToErrorByType[ localDoc.type ] = 
				( self._docsFilteredDueToErrorByType[ localDoc.type ] or 0 ) + 1	
		end
		
		return				
	end

	-- If no existing Oz doc, generate a create request.

	if not ozDoc then
		trackDocumentForDiagnostics( self, 'create', localDoc )
		return deltaForCreateDoc( self, localDoc, newDocsFilteredDueToOzError )
	end

	-- If existing Oz doc and local doc are at the same revision id,
	-- omit creating the overlay

	if ozDoc and ozDoc._revId == localDoc._revId then
		return
	end

	-- Document exists both locally and on Oz. Generate update delta.

	trackDocumentForDiagnostics( self, 'update', localDoc )
	
	return deltaForUpdateDoc( self, localDoc, ozDoc, newDocsFilteredDueToOzError )

end

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

local function convertChangedDocsAndOzDocsToDeltas( self, changedDocs, ozDocs )

	local docIdsToPosition = {}
	local newDocsFilteredDueToOzError = {}
	local result = {}
	local nResults = 0

	local ozDocsIndex = 1
	local nOzDocs = #ozDocs
	
	for i, localDoc in ipairs( changedDocs ) do

		local ozDoc

		if ozDocsIndex <= nOzDocs and ozDocs[ ozDocsIndex ]._id == localDoc._id then
			ozDoc = ozDocs[ ozDocsIndex ]
			ozDocsIndex = ozDocsIndex + 1
		end

		logger:debugf( "convertOneDocToDelta\n\nlocalDoc = %s\n\nozDoc = %s\n\n",
					   localDoc, ozDoc )

		local delta, extraTrashDelta = convertOneDocToDelta( self, localDoc, ozDoc, newDocsFilteredDueToOzError )

		if delta then
		
			if self._isSpacePusher then
			
				delta.space_id = localDoc._spaceId or nil
				assert( delta.space_id )

			end

			if delta.payload and next( delta.payload ) == nil then
				delta.payload = nil
			end
		
		end
		
		logger:infof( "delta for %s = %s\n\n", localDoc._id, delta )

		local function addDelta( delta )

			if delta then

				nResults = nResults + 1
				result[ nResults ] = delta
				docIdsToPosition[ localDoc._localId ] = nResults

			end

		end

		-- Unfortunately oz is not able to create a trash asset
		-- from another trash asset, or restore from a regular non-trash asset.
		-- The createDoc for trashed and restored docs recognizes this and
		-- adds the extra delta needed to make the intermediate document.
		addDelta( extraTrashDelta )

		addDelta( delta )

	end

	return result, docIdsToPosition, newDocsFilteredDueToOzError

end

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

local function makeConvertChangedDocsAndOzDocsToDeltasFn( self )
 
	return function( ... )
		local deltasToUpload, docIdsToDeltaPosition, docsFilteredDueToOzError = convertChangedDocsAndOzDocsToDeltas( self, ... )
		
		-- Mark any docs that were filtered out because we expect to hit StorageExceededError
		-- or TrialOrSubscriptionExpiredError with the appropriate error without sending it to Oz at 
		-- all. Even if this is done prematurely and Oz would have taken it, the logic that retries
		-- these should see these and send them if appropriate.
		if next( docsFilteredDueToOzError ) then
			
			-- NMB-6496: Need to look up dependent docs for these docs and mark them as well to minimize
			-- the number of errors Oz may have to deal with. The majority of these would be album_asset 
			-- records so identify those.
			
			-- sort the table of docs filtered out into some tables we can use to add dependent docs
			local allFilteredDocIds = {}
			local filteredDocIdsToErrorMap = {}
			for errorStr, docIds in pairs( docsFilteredDueToOzError ) do
				for _, docId in ipairs( docIds ) do
					allFilteredDocIds[ #allFilteredDocIds + 1 ] = docId
					filteredDocIdsToErrorMap[ docId ] = errorStr
					self._docsFilteredDueToErrorByType[ 'asset' ] = ( self._docsFilteredDueToErrorByType[ 'asset' ] or 0  ) + 1
				end
			end
			
			if not self._albumAssetsForFilteredAssetsQueryFn then
				-- in order to filter out album_asset records for assets clients should provide an appropriate query function when creating
				-- the WFCatMetadataPusher that will return the album_asset records that correspond to the asset ids given
				logger:warnf( "Assets filtered due to errors, but no _albumAssetsForFilteredAssetsQueryFn to filter out associated album_assets" )

				return WFCatOzErrors.markDocsWithOzError( self._catalog:docStore(), docsFilteredDueToOzError, 'prefiltered' )
					:takeLast( 1 )
					:map( function()
						return deltasToUpload
					end )

			else
			
				return self._albumAssetsForFilteredAssetsQueryFn( self._catalog, allFilteredDocIds )
					:take( 1 )
					:flatMap( function( albumAssetRows ) 
						
						for _, row in ipairs( albumAssetRows ) do
							local albumAssetDocId = row[ 1 ]
							local assetId = row[ 2 ]
							
							-- add them to the docsFilteredDueToOzError
							local errorStr = filteredDocIdsToErrorMap[ assetId ]
							local errors = docsFilteredDueToOzError[ errorStr ]
							errors[ #errors + 1 ] = albumAssetDocId
							
							-- check to see if they're in the deltasToUpload, if so, remove them (log this)
							local maybePosition = docIdsToDeltaPosition[ albumAssetDocId ]
							if maybePosition then
								if deltasToUpload[ maybePosition ] and deltasToUpload[ maybePosition ].action == 'create' then
									logger:warnf( "Removing create of document %s which depends on %s from changed docs to upload due to %s", 
													albumAssetDocId, assetId, errorStr )
									table.remove( deltasToUpload, docIdsToDeltaPosition[ albumAssetDocId ] )
									self._docsFilteredDueToErrorByType[ 'album_asset' ] = ( self._docsFilteredDueToErrorByType[ 'album_asset' ] or 0  ) + 1
								end
							end 
								
						end
						
						-- apply the appropriate error annotation to all of these docs
						return WFCatOzErrors.markDocsWithOzError( self._catalog:docStore(), docsFilteredDueToOzError, 'prefiltered' )
							:takeLast( 1 )
							:map( function()
								return deltasToUpload 
							end )
					end )
			end
		end
		
		return Observable.just( deltasToUpload )
	end
	
end

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

local function putAndTrackPerformance( self, bulkPostUrl, json, options, nDeltasToUpload, updateId )

	return WFPerformanceReport.wrapObservable( 'WFCatMetadataPusher : sendUpdatesToOz', 
		self._catalogSpec:put( bulkPostUrl, json, options ),
		{ 
			bulkPostUrl = bulkPostUrl,
			deltasToUpload = nDeltasToUpload,
			updateId = updateId,
		} )
		
end

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

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

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

local function maybeSetRetryDelayRequestByOz( self, headers )

	local retryDelay = retryDelayRequested( headers )

	if retryDelay then

		if self._retryDelayRequestByOz then
			-- Don't think this can happen.
			logger:error( "Trying to add a retryDelay when one is already set, using new delay" )
		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 redactUserDataFromError( errorString )
	-- If there are other instances where we are inadvertently sending user info in the analytics
	-- messages, they can also be redacted here before we're sending data to analytics.
	
	-- WF-896: Some errors caught when accessing the db can include the full path to the db in the 
	-- error message. We don't want to send this to analytics so redact the username.
	errorString = string.gsub( errorString, "/Users/[^/]+", "/Users/USERNAME" )
	errorString = string.gsub( errorString, "\\\\Users\\\\[^\\]+", "\\\\Users\\\\USERNAME" )
	return string.gsub( errorString, "\\Users\\[^\\]+", "\\Users\\USERNAME" )
end

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

local function recordError( bulkPostUrl, rawJsonOrError, headers )
	
	local errorString = redactUserDataFromError( tostring( rawJsonOrError ) or "" )
	
	logger:errorf( "ERROR on send changes request\n\nPUT %s\n\n%s\n\n%s\n\n",
							   bulkPostUrl, headers, errorString )
	
	local errorEvent = {}
	
	errorEvent['event.type'] = 'error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'metadata_pusher'
	errorEvent['event.url'] =  bulkPostUrl
	errorEvent['event.error_type'] = 'MetadataUpload.Error'
	errorEvent['event.error_code'] = headers and headers.status_code or 'Unknown'
	errorEvent['event.error_desc'] =  errorString
	
	WFAnalytics.trackEvent( errorEvent )
end

--------------------------------------------------------------------------------
--[[!
	\brief Sends one batch of updates to Oz.

	\param bufferedChangedDocs (table) A table of document updates that may
		need to be sent to Oz.
		
	\param numRequiredUpdates (optional number) This is only used by unit tests. If
		set, it's the exact number of updates we expect to send to Oz. 

	\return Observable< number | boolean > If any new documents were sent, the
		number of documents sent. If no documents sent, returns true if there
		is new content to send (but none were generated by this batch). Returns
		false if there is nothing new to send.
]]--

local function doSendUpdatesOnce( self, bufferedChangedDocs, numRequiredUpdates )

	check( 'WFCatMetadataPusher', 'table', '?number' )

	local docStore = assert( self._catalog:docStore() )

	local bulkPostUrl = self._isSpacePusher and self._catalogSpec:spaceUploadUrl() 
			or self._catalogSpec:catalogUpdatesUrl()

	local newContinueFrom, nDeltasToUpload, updateId, deltaProcessStartTime
	
	local sendUpdatesObservable = Observable.just( bufferedChangedDocs )

		:flatMap( function( changedDocs )
			
			deltaProcessStartTime = timeNowForAnimation()
			
			updateId = KSIdGenerator.uuidInLowercaseHex()
			
			local originalContinueFrom = changedDocs.continueFrom
			
			local docsWithNoConflictsAndNoOzLabel = {
				continueFrom = changedDocs.continueFrom 
			}
			
			for i, changedDoc in ipairs( changedDocs ) do
			
				if changedDoc then

					if changedDoc._conflict then
						-- doc is in conflict, don't let change counter pass it by
						docsWithNoConflictsAndNoOzLabel.continueFrom = changedDoc._changeCounter
						break
					elseif changedDoc._labels and ( changedDoc._labels[ 'oz' ] or changedDoc._labels[ '_dontsendtooz' ] ) then
						logger:infof( "Skipping document %s which has the label 'oz' or '_dontsendtooz'", changedDoc._id )
						docsWithNoConflictsAndNoOzLabel.continueFrom = changedDoc._changeCounter + 1
					else
						-- doc is ok, change counter should be next document
						docsWithNoConflictsAndNoOzLabel[ #docsWithNoConflictsAndNoOzLabel + 1 ] = changedDoc
						docsWithNoConflictsAndNoOzLabel.continueFrom = changedDoc._changeCounter + 1
					end
				
				end
				
			end

			changedDocs = docsWithNoConflictsAndNoOzLabel 
			
			-- TODO [scouten 2015-08-05]: Possible race condition?
			-- What if Oz docs gets updated in the interval between
			-- :changedDocs and the fetch of Oz docs here?

			newContinueFrom = changedDocs.continueFrom

			logger:infof( "#changedDocs = %s, originalContinueFrom = %s, newContinueFrom = %s",
						   #changedDocs, originalContinueFrom, newContinueFrom )

			if #changedDocs > 0 then
				return combineChangedDocsWithOz( docStore, changedDocs )
			else
				return Observable.just( changedDocs, {} )
			end

		end )
				
		-- Log WF-flavored documents.

		:tap( function( changedDocs, ozDocs )

			if type( changedDocs ) == 'table' and #changedDocs > 0 then
				logger:debugf( "Changed docs returned \n\n%s\n\n", changedDocs )
			else
				logger:debug( "No documents to send" )
			end

		end )

		-- Convert the local-doc / Oz-doc pairs into deltas.

		:flatMap( makeConvertChangedDocsAndOzDocsToDeltasFn( self ) )
		
		-- SPAM [jandrew 2017-10-11]: Even though the following `flatMap` doesn't change 
		-- the data, it's important that we use `flatMap` and not `tap` because we need
		-- _isDoingWork to fire before we start uploading metadata changes.
		-- This is because `tap` is not guaranteed to execute before the chain 
		-- completes.
			
		:flatMap( function( deltasToUpload )

			if numRequiredUpdates and #deltasToUpload < numRequiredUpdates then

				-- This numRequiredUpdates is only used by unit tests calling _DEBUG_sendNUpdates.

				logger:warnf( "sendUpdatesOnce numUpdates %s < numRequiredUpdates %s",
					#deltasToUpload, numRequiredUpdates )

				return Observable.throw( kNotEnoughUpdatesError )

			end

			if type( deltasToUpload ) == 'table' and #deltasToUpload > 0 then
			
				self._isDoingWork:onNext( true )
				logger:debugf( "Will upload deltas \n\n%s\n\n", deltasToUpload )
				
			end

			return Observable.just( deltasToUpload )

		end )

		-- Upload those documents via bulk doc API.

		:takeLast( 1 )

		:flatMap( function( deltasToUpload )

			self._sendUpdateDeltaProcessTime = timeNowForAnimation() - deltaProcessStartTime

			nDeltasToUpload = #deltasToUpload

			if nDeltasToUpload > 0 then
				
				local json = KSJsonEncoder.encodeOzUploadJson( deltasToUpload )

				local options = {
					requestHeaders = {
						{ header = 'Content-Type', value = 'application/json' },
						{ header = 'Accept', value = 'application/json' },
					},
					requestMethod = 'POST',
					bypassRateLimiter = true,
				}
				
				if WFConfiguration.WFCatMetadataPusher.QuickerRetries then
					table.insert( options.requestHeaders, { header = 'X-Quicker-Retries', value = 'true' } )
				end

				logger:debugf( "Sending docs via\n\nPOST %s\n\n%s\n\n%s\n\n",
							   bulkPostUrl, options, json )

				return maybeAddDelayRequestByOz( self )
				:flatMap( function()
					return putAndTrackPerformance( self, bulkPostUrl, json, options, nDeltasToUpload, updateId )
				end )
				:tap(function()
					self._totalDocumentsSent = self._totalDocumentsSent + #deltasToUpload
					self._lastUpdateCycleWithData = KSDate.timeInIso8601FormatWithTimeZone()
					self._successfulUpdateCycles = self._successfulUpdateCycles + 1
				end)
				
			else

				return Observable.just( true )

			end

		end )

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

		:tap( function( rawJson, headers )

			maybeSetRetryDelayRequestByOz( self, headers )

			if rawJson then
				logger:debugf( "Send docs\n\nPOST %s\n\n%s\n\n%s\n\n",
							   bulkPostUrl, headers, rawJson )
			end

		end )

		-- Record where we left off, so we don't repeat ourselves next time.

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

		:flatMap( function()

			logger:infof( "New continueFrom => %s", newContinueFrom )

			if newContinueFrom then

				self._sendUpdatesContinueFrom:onNext( newContinueFrom )
				
				return docStore:setLocalDocValue {
					_id = self._continueFromLocalDocValueKey,
					continueFrom = newContinueFrom,
				}
				
				:concat( Observable.just( true ) )

			else

				return Observable.just( false )

			end

		end )

		:map( function()

			-- Return true if new content; false if not. This is only
			-- used by unit tests.

			return nDeltasToUpload or not not newContinueFrom
			
		end )
	
	local kSendMetadataFailure = 'FailedToSendUpdateToOz'
	
	return WFRxRetryObservable
		{
			maxAttempts = 100000,
				-- We can't use math.huge here, but the value is large enough that it's essentially a
				-- limitless retry loop.
			progressiveBackoffDelay = 5,
			maximumRetryDelay = 60,
			factory = function()
			
				return sendUpdatesObservable
				
					:catch( function( rawJsonOrError, headers )
			
						recordError( bulkPostUrl, rawJsonOrError, headers )
					
						if not self._syncIsStopped and self._pausedBySynchronizedOperation == 0 then

							logger:debug( "Metadata Push failed. Will retry metadata push after delay." )

							return Observable.throw( rawJsonOrError, headers )

						else
						
							if self._syncIsStopped then
								logger:debug( "Not running sendUpdatesOnce as sync is stopped" )
							elseif self._pausedBySynchronizedOperation > 0 then
								logger:debug( "Not running sendUpdatesOnce as sync is pausedBySynchronizedOperation" )
							else
								logger:debug( "sendUpdatesOnce has completed and is now idle." )
							end
							
							self._isDoingWork:onNext( false )

							-- When we detect that we no longer want to retry, we must return an observable
							-- within this `catch` block and not throw. However, we want the information
							-- about this error to percolate down the reactive chain so we prepend
							-- the kSendMetadataFailure identifier to the return values. When the code
							-- down stream detects this identifier it will rethrow the error
							-- outside of this WFRxRetryObservable, causing the error to flow down
							-- the Rx chain.
							
							return Observable.just( kSendMetadataFailure, rawJsonOrError, headers )

						end
						
					end )
				
			end
		}
			
		:flatMap( function( ... )
		
			-- If this batch of updates failed to send to Oz, rethrow the error so that
			-- the continous task that drives the updates is forced to restart.
			
			local maybeSendMetadataFailure, maybeRawJsonOrError, maybeHeaders = ...
		
			if maybeSendMetadataFailure == kSendMetadataFailure then
			
				return Observable.throw( maybeRawJsonOrError, maybeHeaders )
				
			end
		
			self._continuousPushTask:taskDidWork()
			self._isDoingWork:onNext( false )
		
			return Observable.just( ... )
		
		end )
	
end

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

-- This function behaves a lot like the Rx operator :bufferWithTimeOrCount, but
-- it contains special handling which will elide stale documents when newer
-- revisions of a document are encountered.
--
-- The observable returned from this function will fire with the buffered documents
-- either when the time window has been exceeded (bufferTimeWindow) or when the
-- buffer is full (maxBufferSize).
--
-- The function also strips out space docs for the catalog pusher, and catalog docs
-- for the space pusher.

local function makeDocValueAwareBufferWithTimeOrCountStream( docValueStream, bufferTimeWindow, maxBufferSize, isSpacePusher )

	return Rx.Observable.create( function( observer )

		local uniqueDocumentCount = 0
		local seenDocumentIds = {}
		local currentBufferedValues = {}
		local havePendingTimer
		local largestSeenChangeCounter
		
		local scheduler = WFRxScheduler.sharedScheduler()
		local currTimerSubscr = Rx.SerialDisposable()
			-- holds a "most recent" timer that can be killed via :dispose()
			
		local sendUpdateAndResetLocalState = function()
		
			if #currentBufferedValues > 0 and largestSeenChangeCounter then
		
				local thisIntervalBuffer = currentBufferedValues
				thisIntervalBuffer.continueFrom = largestSeenChangeCounter + 1
				
				uniqueDocumentCount = 0
				seenDocumentIds = {}
				currentBufferedValues = {}
				havePendingTimer = false
				largestSeenChangeCounter = nil
				
				observer:onNext( thisIntervalBuffer )
			
			end
			
		end
		
		local previousDocValueInConflict = false
		
		local valueCollectorSubscr =
			docValueStream
			
				:takeWhile( function( docValue )
		
					-- When we detect the conflict at sequence N, we stop listening for additional doc value changes.
					-- We will then send documents M..N to doSendUpdatesOnce, where we will then push documents 
					-- M..N-1 to oz. Afterwhich the continuous task that drives the pusher will complete.
					-- It will naturally be restarted after a few seconds using sequence number of N as the
					-- starting point. After restarting, if the document is no longer in conflict, the task
					-- will be able to push all updates >= N. If the document is still in conflict, then the task
					-- will complete, wait, and then restart until the document is no longer in conflict.
					
					return not previousDocValueInConflict
				
				end )
			
				:subscribe(
					function( docValue )
					
						previousDocValueInConflict = docValue._conflict
					
						if docValue then
						
							-- For the space pusher, we only care about documents with space IDs.
							-- For the catalog pusher, we only care about documents without space IDs.
							-- Note that this filtering happens in doSendUpdatesOnce so that the
							-- tracked change counter is still incremented even if we can filter
							-- out a particular changedDoc.

							local isUpdateRelevant = ( not isSpacePusher and not docValue._spaceId ) or
									( isSpacePusher and docValue._spaceId )
						
							if isUpdateRelevant then
						
								-- Keep track of which document Ids exist in the current
								-- set of buffered documents.
							
								local docValueId = docValue._id
								local docValueIndex = seenDocumentIds[ docValueId ]

								if docValueIndex then
								
									-- If a newer revision of document is detected, then
									-- remove the stale revision of the document from the
									-- current set of buffered docs. 
								
									currentBufferedValues[ docValueIndex ] = false
									
								else
									uniqueDocumentCount = uniqueDocumentCount + 1
								end

								docValueIndex = #currentBufferedValues + 1
								seenDocumentIds[ docValueId ] = docValueIndex
								currentBufferedValues[ docValueIndex ] = docValue

							else
							
								currentBufferedValues[ #currentBufferedValues + 1 ] = false
							
							end

							local changeCounter = docValue._changeCounter

							if not largestSeenChangeCounter or 
									( changeCounter and changeCounter > largestSeenChangeCounter ) then
									
								-- This code will extracts out the highest changeCounter
								-- in the current batch of buffered documents and makes it a
								-- top-level field in the resulting table. This is used by
								-- the doSendUpdates fn.
									
								largestSeenChangeCounter = changeCounter
								
							end

						end

						if uniqueDocumentCount == maxBufferSize then
						
							-- Once the buffer is full, send the batch of documents downstream!
						
							sendUpdateAndResetLocalState()

						elseif not havePendingTimer then

							-- Once the time window has elapsed, send the batch of documents downstream!

							havePendingTimer = true

							currTimerSubscr
								:setDisposable(
									scheduler
										:scheduleFuture( 
											bufferTimeWindow,
											sendUpdateAndResetLocalState ) )

						end

					end,
					function( ... )
					
						-- If docValueStream throws an error, attempt to send whatever
						-- documents we've seen up to this point.
						
						sendUpdateAndResetLocalState() 
					
						observer:onError( ... ) -- relay errors
						
					end,
					function()
					
						-- If docValueStream completed, this could be the result of a conflict
						-- being detected in the takeWhile section above. Attempt to send whatever
						-- documents we've seen up to this point.
						
						sendUpdateAndResetLocalState() 

						observer:onCompleted() -- relay completion
						
					end )

		return Rx.CompositeDisposable(
				currTimerSubscr, valueCollectorSubscr )

	end )

end

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

local function debug_makeDocValueAwareBufferWithTimeOrCountStream( _, ... )

	return makeDocValueAwareBufferWithTimeOrCountStream( ... )

end

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

-- This function is required by the call to scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions
-- below. Right now, this function does not filter out any updates; every single
-- update is passed through. Though we could filter out documents here based on
-- the presence of an oz label or conflicts, this would require that we change the
-- way that we update the continue-from change counter associated with the pusher.

local function reportAllChanges( _, scanState, docValue )

	if docValue then
		scanState.SFMCFSUpdateCount = scanState.SFMCFSUpdateCount + 1
	end
	
	return scanState, docValue, scanState.SFMCFSUpdateCount
	
end

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

local function sendUpdatesTask( self, debug_CalledOutsideOfContinuousTask, debug_NumRequiredUpdates )

	local previousSFMCFSUpdateCount = 0
	local previousDocValueChangeCounter = -1
	
	local docValueStream = self._sendUpdatesContinueFrom
		
		:take( 1 )
		
		:flatMap( function( continueFromChangeCounter )
		
			-- Return a stream of document updates from the doc store, starting
			-- with the change counter of the last document we pushed to Oz + 1.
			
			return self._catalog:docStore()
				:scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions( 
						'WFCatMetadataPusher', 
						'reportAllChanges', 
						{
							SFMCFSUpdateCount = 0
						},
						continueFromChangeCounter )
			
		end )

		:flatMap( function( docValue, SFMCFSUpdateCount )

			-- SPAM [jandrew 2021-02-02]: The following condition was added to address a sync issue
			-- that was encountered via WF-1452. As of writing this there is no strong evidence 
			-- that scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions can drop updates or deliver updates
			-- out of order.
			
			-- SFMCFSUpdateCount is a count of all the document updates that have passed through
			-- the WFCatMetadataPusher.reportAllChanges() function. If ever we detect that this
			-- value has incremented by more than one, it means that scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions
			-- failed to send us a document update, or that the document update was delivered
			-- out of order. In either of these scenarios, the safest thing to do is to
			-- throw an error which will cause the sendUpdatesTask to be restarted.

			if SFMCFSUpdateCount ~= previousSFMCFSUpdateCount + 1 then
			
				local numberOfDroppedUpdates = SFMCFSUpdateCount - previousSFMCFSUpdateCount - 1
				local errorMessage = string.format( "scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions may have failed to send us %d doc updates.", numberOfDroppedUpdates )
				
				logger:warn( errorMessage )
			
				return Observable.throw( errorMessage )
			
			elseif docValue._changeCounter < previousDocValueChangeCounter then
			
				-- Warn but don't throw if documents are received out of order via changeCounter. This shouldn't 
				-- happen, but if it happens as a result of a bug within scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions, 
				-- restarting sendUpdatesTask may not fix the issue.
				
				local errorMessage = "scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions sent document updates out of order."
				
				logger:warn( errorMessage )
				
			end
			
			previousSFMCFSUpdateCount = SFMCFSUpdateCount
			previousDocValueChangeCounter = docValue._changeCounter
			
			if debug_CalledOutsideOfContinuousTask then
			
				return Observable.just( docValue )
					-- Unit tests can bypass the continuous task to call this function.
					
			else
			
				return self._continuousPushTask:throwIfNotRunning( docValue )
					-- The happy path.
					-- Need to throw as this function doesn't ever complete.
					
			end
				
		end )

	return makeDocValueAwareBufferWithTimeOrCountStream( 
			docValueStream, self._delayBeforeSendingUpdates, self._documentBatchCount, self._isSpacePusher )
		
		:filter( function( bufferedChangedDocs )
		
			return bufferedChangedDocs and #bufferedChangedDocs > 0
			
		end )
				
		:map( function( bufferedChangedDocs )
		
			-- Attempt to send a batch of updates to Oz.
		
			return doSendUpdatesOnce( self, bufferedChangedDocs, debug_NumRequiredUpdates )
					
		end )
		
		:flatMerge( 1 ) 
			-- Make sure only one batch of updates processes at a time.
		
		:catch( function( ... )

			if debug_CalledOutsideOfContinuousTask then

				return Observable.throw( ... )

			else

				return self._continuousPushTask:throwIfNotStopped( ... )

			end

		end )

end


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

local WFCatMetadataPusher = {

	_catalog = false,
		-- (WFCatalog)

	_catalogSpec = false,
		-- (WFCatSpec)

	_sendUpdateDeltaProcessTime = false,
		-- (number) Process time for delta. Including docStore time and local calculation.

	_sendUpdatesContinueFrom = false,
		-- (ReplaySubject< number >)

	_continuousPushTask = false,
		-- (WFRxContinuousTask) A continuous task to push metadata updates to Oz.

	_requestHeaders = false,
		-- (table) HTTP Request Headers to be used for bulk upload

	_isDoingWork = false,
		-- (Rx ReplaySubject) A Rx.Subject used to indicate wether not uploading is occuring.
	
	_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) internal flag used to indicate whether sync should be stopped.

	_pausedBySynchronizedOperation = 0,
		-- (number) internal flag used to indiate sync is paused (> 0) because of a synchronized operation.

	_userIsOverQuota = false,
		-- (boolean) internal flag used to indicate whether user is over quota and we should modify how we handle
		-- asset creates on Oz
	
	_userIsExpired = false,
		-- (boolean) internal flag used to indicate whether user's trial or subscription is expired and we should modify
		-- how we handle asset creates on Oz
		
	_delayBeforeSendingUpdates = false,
		-- (number) Delay in seconds the pusher waits to collect batches before sending new updates
		
	_documentBatchCount = false,
		-- (number) Number of documents to batch before uploading to Oz
		
	_createDocRemapFunctions = false,
		-- (table) Remap functions applied during document creation on Oz
		
	_updateDocRemapFunctions = false,
		-- (table) Remap functions applied during document updates on Oz
	
	_albumAssetsForFilteredAssetsQueryFn = false,
		-- (function) A function that takes a catalog and a list of full asset ids and returns an Observable that
		-- will query for the album_assets associated with these asset ids.  The rows returned by this observable should contain
		-- the album_asset docId first and the full *asset* id from the album_asset record.  e.g.
		-- SELECT album_asset_v2.docId, album_asset_v2.assetId FROM CARRAY( ?, ?, 'char*' ) as ai JOIN album_asset_v2 ON 
		--					ai.value = album_asset_v2.assetId
		--  albumIds,
		--  #albumIds,

	_isSpacePusher = false,
		-- (boolean) if true, this pusher will only process documents with spaceIds.
		-- if false, this pusher will only process catalog-based documents.

	_continueFromLocalDocValueKey = false,
		-- (string) The local doc value key used by this pusher to track continue-from.

	_spacesOverQuotaById = false,
		-- (table) A table used track which spaces are over quota.
		-- key: space ID
		-- value: true if the space is over quota.
		
	_spacesExpiredById = false,
		-- (table) A table used track which spaces have lapsed subscriptions.
		-- key: space ID
		-- value: true if the space owner has a lapsed subscription.
		
	-- fields storing data for the diagnostic log

	_docsFilteredDueToErrorByType = {},
		-- (table) count of uploads filtered due to preexisting error by document type
		
	_successfulUpdateCycles = 0,
		-- (number) total number of successful update cycles since app launch
		
	_totalDocumentsSent = 0,
		-- (number) total number of documents sent since app launch
		
	_lastUpdateCycleWithData = false,
		-- (string) timestamp of the last update sent to Oz that contained data

	_createDocTypeAllowlist = false,
		-- (table) An allowlist of documents that can be created on Oz.
		
	_updateDocTypeAllowlist = false,
		-- (table) An allowlist of documents that can be updated on Oz.
		
	_deleteDocTypeAllowlist = false,
		-- (table) An allowlist of documents that can be deleted from Oz.
		
	_documentsSentByTypeStream = false,
		-- (Rx BehaviorSubject)

	_documentsSentByType = false,
		-- (table) table with counts of documents that could be sent by document type

	_recentlyPushedDocIds = false,
		-- (WFRingBuffer)

	class = {
		getContinueFromLocalDocValueKey = getContinueFromLocalDocValueKey,
		reportAllChanges = reportAllChanges,
		makeDocValueAwareBufferWithTimeOrCountStream = debug_makeDocValueAwareBufferWithTimeOrCountStream
			-- This is only exposed for testing purposes.
	},
		
	WFCloseSupport.addCloseSupportMethods,

}

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

function WFCatMetadataPusher:__init( catalog, options, optCatSpec )

	check( 'WFCatMetadataPusher', 'WFCatalog', '?table', '?WFCatSpec'  )

	self:setOwned {
		_catalog = catalog:relayObject()
	}
	
	options = options or {}
	
	self._createDocRemapFunctions = {}
	self._updateDocRemapFunctions = {}
	
	self._syncIsStopped = true
	self._pausedBySynchronizedOperation = 0

	self._catalogSpec = optCatSpec or catalog:ozCatalogSpec()

	self._sendUpdatesContinueFrom = ReplaySubject( 1 )
	
	self._delayBeforeSendingUpdates = options.delayBeforeSendingUpdates 
			or WFPlatformInfo.isUnitTest() and 0.1
			or 3
	
	self._documentBatchCount = options.documentBatchCount or 100
	self._albumAssetsForFilteredAssetsQueryFn = options.albumAssetsForFilteredAssetsQueryFn
	self._isSpacePusher = not not options.isSpacePusher
	
	self._createDocTypeAllowlist = createDocTypeAllowlist
	self._updateDocTypeAllowlist = updateDocTypeAllowlist
	self._deleteDocTypeAllowlist = deleteDocTypeAllowlist
	
	if self._isSpacePusher then
	
		-- For testing, we allow the standard set of documents to be used instead of the
		-- more restrictive space-feed set. This allows us to write more thorough
		-- unit tests. If useSpaceFeedDocTypeAllowList is set, we will use the restricted
		-- set of document types.
		
		if options.useSpaceFeedDocTypeAllowList then
		
			self._createDocTypeAllowlist = spaceFeedCreateDocTypeAllowlist
			self._updateDocTypeAllowlist = spaceFeedUpdateDocTypeAllowlist
			self._deleteDocTypeAllowlist = spaceFeedDeleteDocTypeAllowlist
			
		end
		
	end
	
	self._continueFromLocalDocValueKey = getContinueFromLocalDocValueKey( nil, self._isSpacePusher )
	
	self._spacesOverQuotaById = {}
	
	self._spacesExpiredById = {}
	
	self._documentsSentByType = {}
	
	WFDiagnosticLog.addLogProvider( self )
	
	self._requestHeaders = {
		{ header = 'Content-Type', value = 'application/json' },
		{ header = 'Accept', value = 'application/json' },
	}

	updateSendUpdatesContinueFrom( self )
		:call( runUntilDone )

	local continuousTaskDelays = options.continuousTaskDelays or {
	
		progressiveBackoff = true,
		errorDelay = 10,
		maxDelay = 60,

	}
	
	local continuousTask = function() return sendUpdatesTask( self ) end

	self._continuousPushTask = WFRxContinuousTask( "WFCatMetadataPusher_"..catalog:ozCatalogId(), continuousTask, continuousTaskDelays )

	self._isDoingWork = ReplaySubject( 1 )
	self._isDoingWork:onNext( false )
	
	self:addCreateDocRemapFunction( 'album_asset', album_assetCreateRemapFn )
	self:addCreateDocRemapFunction( 'space_album', space_albumCreateRemapFn )
	self:addCreateDocRemapFunction( 'stack_asset', stack_assetCreateRemapFn )
	self:addCreateDocRemapFunction( 'comment', commentCreateRemapFn )
	self:addCreateDocRemapFunction( 'favorite', favoriteCreateRemapFn )
	self:addCreateDocRemapFunction( 'asset', assetCreateRemapFn )
	self:addCreateDocRemapFunction( 'spaceinvite', spaceinviteCreateRemapFn )
	self:addCreateDocRemapFunction( 'version', versionCreateRemapFn )
	self:addUpdateDocRemapFunction( 'album_asset', album_assetUpdateRemapFn )
	self:addUpdateDocRemapFunction( 'version', versionUpdateRemapFn )
	
end

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

function WFCatMetadataPusher:close()

	check( 'WFCatMetadataPusher' )
	
	self:stopContinuousUpdates()
	
	if self._sendUpdatesContinueFrom then
		self._sendUpdatesContinueFrom:dispose()
		self._sendUpdatesContinueFrom = nil
	end

	local continuousPushTask = self._continuousPushTask
	self._continuousPushTask = nil
	
	if continuousPushTask then
		continuousPushTask:close()
	end
	
	local documentsSentByTypeStream = self._documentsSentByTypeStream
	self._documentsSentByTypeStream = nil

	if documentsSentByTypeStream then
		documentsSentByTypeStream:dispose()
	end

end

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

function WFCatMetadataPusher:addCreateDocRemapFunction( docType, remapFn )

	check( 'WFCatMetadataPusher', 'string', 'function' )
	
	if self._createDocRemapFunctions[ docType ] then
		logger:warnf( 'Create Document remap function already defined for docType %s', docType )
	end
	
	self._createDocRemapFunctions[ docType ] = remapFn
	
end

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

function WFCatMetadataPusher:addUpdateDocRemapFunction( docType, remapFn )

	check( 'WFCatMetadataPusher', 'string', 'function' )
	
	if self._updateDocRemapFunctions[ docType ] then
		logger:warnf( 'Update Document remap function already defined for docType %s', docType )
	end
	
	self._updateDocRemapFunctions[ docType ] = remapFn
	
end

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

function WFCatMetadataPusher:addRequestHeader( name, value )

	check( 'WFCatMetadataPusher', 'string', 'string|number|boolean' )
	
	table.insert( self._requestHeaders, { header = name, value = value } )
	
end

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

function WFCatMetadataPusher:removeRequestHeader( name )

	check( 'WFCatMetadataPusher', 'string' )
	
	local WFFunctionUtils = require 'WFFunctionUtils'
	
	self._requestHeaders = WFFunctionUtils.filter( function( request )
	
		if request.header == name then
			return nil
		end
		
		return request
		
	end, self._requestHeaders )
	
end

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

function WFCatMetadataPusher:stopContinuousUpdates()

	check( 'WFCatMetadataPusher' )
	
	self._syncIsStopped = true
	self._continuousPushTask:stopContinuousTask()

end

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

function WFCatMetadataPusher:sendUpdatesContinuously()

	check( 'WFCatMetadataPusher' )
	
	self._syncIsStopped = false
	if self._pausedBySynchronizedOperation == 0 then
		self._continuousPushTask:startContinuousTask()
	end

end

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

function WFCatMetadataPusher:isDoingWork()

	check( 'WFCatMetadataPusher' )

	return self._isDoingWork

end

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

--[[!
	\brief This function returns an observable that will fire with true every time
	the pusher has completed all known work.
]]--

function WFCatMetadataPusher:workHasCompletedStream()

	return self._isDoingWork
	
		:distinctUntilChanged()
	
		:filter( function( isDoingWork )
			return isDoingWork
		end )
		
		:flatMap( function()
		
			return self._isDoingWork
			
				:distinctUntilChanged()
		
				:filter( function( isDoingWork )
					return not isDoingWork
				end )
				
				:take( 1 )
				
				:map( function()
					return true
				end )
		
		end )

end

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

function WFCatMetadataPusher:_DEBUG_sendUpdatesOnce( debugNumRequiredUpdates )

	check( 'WFCatMetadataPusher', '?number' )
	
	local attemptToSendUpdates = sendUpdatesTask( self, true, debugNumRequiredUpdates ):take( 1 )
	
	if debugNumRequiredUpdates == 0 then
	
		return self:countUpdatesToSend()
		
			:take( 1 )
			
			:flatMap( function( numberOfUpdates )
			
				if numberOfUpdates > 0 then
				
					return attemptToSendUpdates
				
				else
				
					return Observable.just( 0 )
				
				end
			
			end )	
	
	end
	
	return attemptToSendUpdates

end

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

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

function WFCatMetadataPusher:_testOnly_getIsTaskRunningStream()

	check( 'WFCatMetadataPusher' )

	return self._continuousPushTask:_testOnly_getIsTaskRunningStream()

end

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

--[[!
	\brief This function is only used in unit tests.

	Calling this function will cause the metadata pusher to continously attempt
	to push updates until it has pushed the number of updates specified by n.

	\param numRequiredUpdates (number) The number of updates to push.
	\param timeoutSeconds (number) The number of seconds at which to give up
		and issue one last request to see what updates were available.
	\param allowMoreThanN (boolean) If true, this function will not throw an
		error if more than n updates are sent.

	\return Observable - Resolves when n updates have been pushed.
]]--

function WFCatMetadataPusher:_DEBUG_sendNUpdates( numRequiredUpdates, timeoutSeconds, allowMoreThanN )

	check( 'WFCatMetadataPusher', 'number', '?number', '?boolean' )

	-- Use one hour as default timeout as, all tests will timeout sooner, all tests run in less than 10 min currently.
	timeoutSeconds = timeoutSeconds or 60*60

	logger:infof( "_DEBUG_sendNUpdates numRequiredUpdates = %s, timeOutSeconds %s, allowMoreThanN = %s",
						  numRequiredUpdates, timeoutSeconds, allowMoreThanN )

	return Observable.defer( function()

		return self:_DEBUG_sendUpdatesOnce( numRequiredUpdates )
		
		:catch( function( errorMessage )

			if errorMessage == kNotEnoughUpdatesError then
				return Observable.throw( errorMessage )
			end

			-- if we threw for some other reason send it on.
			return Observable.just( 0, errorMessage )

		end )

	end )

	:retry( 100000 ) -- can't use math.huge, retrying this many times is unlikley to succeed.

	:timeout( timeoutSeconds, WFRxScheduler.sharedScheduler() )

	:catch( function( errorMessage )

		if errorMessage == "Timeout" then

			-- Delay a little to allow tapOnError in the prior call to
			-- doSendUpdatesOnce to complete prior to re-issuing one
			-- last call to doSendUpdatesOnce.

			return Observable.timer( .1, WFRxScheduler.sharedScheduler() )
			:flatMap( function()

				-- Issue one last call to doSendUpdatesOnce without numRequiredUpdates
				-- to send whatever the pusher has to send to make it easier to
				-- debug as the HttpMock can show more detailed information as to what
				-- is happening.

				return self:_DEBUG_sendUpdatesOnce()

			end )
		
		else
		
			return Observable.throw( errorMessage )
		end

	end )

	:flatMap( function( count, maybeErrorMessage )

		if maybeErrorMessage then
			
			return Observable.throw( maybeErrorMessage )

		elseif count == 0 and count < numRequiredUpdates then

			-- If we got 0 updates and we were expecting more than 0 then show this error
			-- as no http transactions will be generated.  Otherwise send it on as the
			-- error generated by the MockedHttp will be more informative.

			return Observable.throw( string.format(
				"Too few updates were sent. Expected %d, Sent %d", numRequiredUpdates, count ) )

		elseif count > numRequiredUpdates and not allowMoreThanN then

			return Observable.throw( string.format(
				"Too many updates were sent. Expected %d, Sent %d", numRequiredUpdates, count ) )

		end

		return Observable.empty()

	end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Counts the number of document updates that may need to be send to Oz.

	WARNING: This API is prone to over-reporting. Some documents may be marked
	as changed locally, but may wind up being no-ops. (For example, the change
	might have been the read-back of the change \b from Oz.) This API will also
	over report if there exist documents that need to be pushed by a different
	instance of a pusher. (For example, this API may report docs with space IDs 
	that need to be pushed even if this is the Catalog pusher, and not the space
	doc pusher).

	If updates are being sent to Oz, this observable should eventually return
	to the correct result (zero).

	\return Observable< number > number of document updates that may need to be
		sent to Oz
]]--

function WFCatMetadataPusher:countUpdatesToSend()

	check( 'WFCatMetadataPusher' )

	return Observable.combineLatest(

		function( changeCounter, sendUpdatesContinueFrom )

			logger:infof( "countUpdatesToSend changeCounter = %s, sendUpdatesContinueFrom = %s",
						  changeCounter, sendUpdatesContinueFrom )

			if type( changeCounter ) == 'number'
				and type( sendUpdatesContinueFrom ) == 'number'
			then

				if changeCounter >= sendUpdatesContinueFrom then

					-- Yes, the following math is a bit wonky. The continueFrom result
					-- from :changedDocs is inclusive (i.e. the first new revision
					-- number you are seeking). I wish now that it was the change
					-- counter of the last document considered, but fixing that
					-- is much too risky now.

					if sendUpdatesContinueFrom > 0 then
						sendUpdatesContinueFrom = sendUpdatesContinueFrom - 1
					end

					return changeCounter - sendUpdatesContinueFrom

				else

					return 0

				end

			else

				return false

			end

		end,

		self._catalog:docStore():changeCounter(),
		self._sendUpdatesContinueFrom )

		:filter( function( value )
			return value -- discard "false" placeholder that says incomplete data
		end )

end

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

function WFCatMetadataPusher:setUserIsOverQuota( isOverQuota )

	check( 'WFCatMetadataPusher', 'boolean' )
	
	self._userIsOverQuota = isOverQuota
	
end

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

function WFCatMetadataPusher:setUserIsExpired( isExpired )
	
	check( 'WFCatMetadataPusher', 'boolean' )
	
	self._userIsExpired = isExpired
	
end

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

function WFCatMetadataPusher:setSpaceIsOverQuota( spaceId, isOverQuota )

	check( 'WFCatMetadataPusher', '+string', 'boolean' )
	
	self._spacesOverQuotaById[ spaceId ] = isOverQuota
	
end

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

function WFCatMetadataPusher:getSpaceIsOverQuota( spaceId )

	check( 'WFCatMetadataPusher', '+string' )
	
	return self._spacesOverQuotaById[ spaceId ]
	
end

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

function WFCatMetadataPusher:setSpaceIsExpired( spaceId, entitlementExpired )

	check( 'WFCatMetadataPusher', '+string', 'boolean' )
	
	self._spacesExpiredById[ spaceId ] = entitlementExpired
	
end

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

function WFCatMetadataPusher:getSpaceIsExpired( spaceId )

	check( 'WFCatMetadataPusher', '+string' )
	
	return self._spacesExpiredById[ spaceId ]
	
end

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

function WFCatMetadataPusher:pauseReturningCounter( trackWaitTime, trackEventArgs )

	check( 'WFCatMetadataPusher', '?boolean', '?string' )
	
	self._pausedBySynchronizedOperation = self._pausedBySynchronizedOperation + 1

	logger:infof( "pauseReturningCounter _pausedBySynchronizedOperation = %s", self._pausedBySynchronizedOperation )

	self._continuousPushTask:stopContinuousTask()

	local startTime = timeNowForAnimation()
	local kTrackIfTimeOverSecs = 3

	return self._isDoingWork
		:filter( function( isDoingWork )
			return not isDoingWork
		end )
		:flatMap( function()
			local elapsedTime = timeNowForAnimation() - startTime
			
			if trackWaitTime and elapsedTime > kTrackIfTimeOverSecs then
				WFAnalytics.trackEvent {
					['event.type'] = 'pause-pusher',
					['event.subtype'] = trackEventArgs.eventSubtype or nil,
					['event.workflow'] = trackEventArgs.eventWorkflow or 'SYNC',
					['event.subcategory'] = 'metadata_pusher',
					['event.value'] = elapsedTime,
					['event.count'] = self._sendUpdateDeltaProcessTime,
				}
			end
		
			return self._catalog:docStore():localDocValue( self._continueFromLocalDocValueKey )
		end )
		:map( function( docValue )
			return KSTableUtils.derefOpt( docValue, 'continueFrom' ) or 0
		end )
		:take( 1 )

end

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

function WFCatMetadataPusher:resume()

	check( 'WFCatMetadataPusher' )

	self._pausedBySynchronizedOperation = self._pausedBySynchronizedOperation - 1

	logger:infof( "resume _pausedBySynchronizedOperation = %s", self._pausedBySynchronizedOperation )

	if self._pausedBySynchronizedOperation < 0 then
		logger:error( "_pausedBySynchronizedOperation < 0" )
		self._pausedBySynchronizedOperation = 0
	end

	if not self._syncIsStopped and self._pausedBySynchronizedOperation == 0 then
		self._continuousPushTask:startContinuousTask()
	end

end

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

local function createDiagnosticSummaryStr( description, documentsSentByType )

	local numberOfCreatesSynced 
			= derefOpt( documentsSentByType, 'create', 'total' ) or 0
	local numberOfUpdatesSynced 
			= derefOpt( documentsSentByType, 'update', 'total' ) or 0
	local numberOfDeletionsSynced  
			= derefOpt( documentsSentByType, 'delete', 'total' ) or 0

	return string.format( "%s synced up %d new docs, %d updates, %d deletions to Oz.", 
			description, numberOfCreatesSynced, numberOfUpdatesSynced, numberOfDeletionsSynced )
			
end

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

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

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

--[[!
	\brief This API that can be used gather real-time diagnostic data from the
		metadata pusher.
		
	\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.
		
		Please note that the documentsSyncedByType section may include documents that
		were processed, but not synced because it wasn't necessary to send the update to Oz.
	
		Also note that the recentlySyncedDocIds section only shows document Ids that have
		been synced _after_ calling the syncDiagnostics API.
	
	\return Observable< table > An Observable whose value tracks a table of
		sync diagnostic data. An example data event may look like this:
		
		{
			docsFilteredDueToErrorByType = {},
			documentsSyncedByType = {
				create = {
					album = {
						collection = 2,
					},
					album_asset = {
						NULL = 2,
					},
					asset = {
						deleted_image = 2,
					},
					total = 6,
				},
				delete = {
					album = {
						collection = 1,
					},
					error = {
						ResourceExistsError = 1,
					},
					total = 2,
				},
				update = {
					album = {
						collection = 1,
					},
					asset = {
						image = 8,
						video = 1,
					},
					total = 10,
				},
			},
			isCurrentlySyncing = false,
			pausedBySynchronizedOperation = 0,
			recentlySyncedDocIds = {
				"fab34b99fb484df1b464c4c907302dc6",
				"6cfae0f5276b4d2395ecc8ef699eefdb",
				"c1c55a19e64342a5aeab26a40d8fe4bb",
				"4b69dcb505734b12bfa4be8e11d24cfc",
				"e9bff65c294042aa8eb3424842be8a41",
				"33ade424aa1c466684852ea7d05f95a5",
				"614d11910684498398b4d8db8cfa5c4d",
				"12a6e18382a6495b970c5c62b29a746a",
				"950f8a72cb8340149338648a1dd40213",
				"920a5f6d73df4b8b9b9d28b1735aa6ce",
				"7d17745bbffd41258d23c46be59816a4",
				"8c252dae220a41988bd7901fc75e0873",
				"eeab8e3a749141d789a8bf9f097e057d",
				"73d78e05af75452781f2162d50c48ccc",
				"eeab8e3a749141d789a8bf9f097e057d",
				"eeab8e3a749141d789a8bf9f097e057d",
				"8dc28a78b4d1765d04910c808a9efd6a",
				"eeab8e3a749141d789a8bf9f097e057d",
			},
			summaryStr = "Catalog Pusher synced up 6 new docs, 10 updates, 2 deletions to Oz.",
			totalDocumentsSynced = 17,
			numberOfDocsRemainingToSync = 2,
		}
--]]

function WFCatMetadataPusher:syncDiagnostics( optCalm, optBufferSize )

	check( 'WFCatMetadataPusher', '?number', '?number' )

	if not self._documentsSentByTypeStream then
		self._documentsSentByTypeStream = BehaviorSubject( {} )
	end
	
	if not self._recentlyPushedDocIds then
		self._recentlyPushedDocIds = WFRingBuffer.make( optBufferSize or 100 )
	end
	
	local numberOfDocsRemainingToSyncStream
	
	if WFPlatformInfo.isDesktop then
	
		local syncStatusColumnName = WFCatSyncStatus.getSyncStatusColumnName()
	
		numberOfDocsRemainingToSyncStream = self._catalog:indexStore():query( {
				sql = string.format( [[SELECT COUNT( * ) FROM asset_sync_status_ui WHERE %s == 1]], syncStatusColumnName ),
				calm = optCalm or 1,
				singleColumnQuery = true,
			} )
			
			:map( function( rows )
			
				return derefOpt( rows, 1 ) or 0
			
			end )

	else
	
		numberOfDocsRemainingToSyncStream = self._catalog:indexStore():query( {
				sql = [[SELECT COUNT( * ) FROM coreInfo WHERE syncStatus == 1]],
				calm = optCalm or 1,
				singleColumnQuery = true,
			} )
			
			:map( function( rows )
			
				return derefOpt( rows, 1 ) or 0
			
			end )

	end

	local descriptor = self._isSpacePusher and 'Space Pusher' or 'Catalog Pusher'
	
	return Observable.combineLatest(
			self._documentsSentByTypeStream:startWith( {} ),
			self:isDoingWork():startWith( false ),
			numberOfDocsRemainingToSyncStream:startWith( 0 ) )

		:calm( optCalm or 1, WFRxScheduler.sharedScheduler() )

		:map( function( documentsSentByType, isCurrentlySyncing, numberOfDocsRemainingToSync )
		
			local diagnostics = {
				summaryStr = createDiagnosticSummaryStr( descriptor, documentsSentByType ),
				totalDocumentsSynced = self._totalDocumentsSent,
				isCurrentlySyncing = not not isCurrentlySyncing,
				documentsSyncedByType = documentsSentByType,
				recentlySyncedDocIds = WFRingBuffer.map( justReturn, self._recentlyPushedDocIds ),
				pausedBySynchronizedOperation = self._pausedBySynchronizedOperation,
				docsFilteredDueToErrorByType = self._docsFilteredDueToErrorByType,
				numberOfDocsRemainingToSync = numberOfDocsRemainingToSync
			}
			
			logger:infof( "%s syncDiagnostics update: %s", descriptor, diagnostics )
			
			return diagnostics
				
		end )

end

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

function WFCatMetadataPusher:sendDiagnosticLogData()

	check( 'WFCatMetadataPusher' )

	return {
		
		name = 'Metadata Sync Up: ' .. self._catalogSpec:catalogId(),
		id = 'WFCatMetadataPusher-' .. self._catalogSpec:catalogId(),
		weight = 500,
		parentSection = WFDiagnosticLog.Sections.SYNC,

		body = {
			isCurrentlySyncing = not self._syncIsStopped,
			pausedBySynchronizedOperation = self._pausedBySynchronizedOperation,
			changedDocsToSendByType = self._documentsSentByType,
			docsFilteredDueToErrorByType = self._docsFilteredDueToErrorByType,
			successfulUpdateCycles = self._successfulUpdateCycles,
			totalDocumentsSent = self._totalDocumentsSent,
			lastUpdateCycleWithData = self._lastUpdateCycleWithData,
			recentlySyncedDocIds = self._recentlyPushedDocIds and WFRingBuffer.map( justReturn, self._recentlyPushedDocIds ) or nil,
		}
		
	}

end

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

function WFCatMetadataPusher:DEBUG_convertOneDocToDelta( localDoc, ozDoc, newDocsFilteredDueToOzError )
	-- Only used by unit tests.
	newDocsFilteredDueToOzError = newDocsFilteredDueToOzError or {}
	return convertOneDocToDelta( self, localDoc, ozDoc, newDocsFilteredDueToOzError )
end

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

return Rx.Class 'WFCatMetadataPusher'( WFCatMetadataPusher )
