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

WFCatSyncAutoDeleter.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2016 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

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

	-- WF.core.utils
local KSFunctionUtils = require 'KSFunctionUtils'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'

	-- WF.oz_client.managed_catalog
local WFCatDelete = require 'WFCatDelete'
local WFCatSpacePurgeUtils = require 'WFCatSpacePurgeUtils'

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

	-- WF.sqlite.database
local WFSQLiteUtils = require 'WFSQLiteUtils'

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

local logger = WFLogger( 'WFCatSyncAutoDeleter' )

local check = assert( WFArguments.check )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

local runUntilDone = WFRxRunUntilDone

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

local kAutoDeleterChangeCounterLocalDocId = 'WFCatSyncAutoDeleter-changeCounter'

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

-- @mark WFCatSyncAutoDeleter class

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

local WFCatSyncAutoDeleter = {

	_wfCatalog = false,
		-- (WFCatalog)

	_docStore = false,
		-- (WFRxDocStore)
	
	_indexStore = false,
		-- (WFRxIndexStore)
		
	_dataCache = false,
		-- (KSDataCache) KSDataCache database
		-- Only used by Android

	_catalogDocIdsToProcessByDocType = false,
		-- (table)

	_spaceFeedDocIdsToProcessByDocType = false,
		-- (table)
		
	_trashedDocIdsToProcessByDocType = false,
		-- (table)
		
	_deletedDocIdStreamSubscription = false,
		-- (Rx Subscription)
		
	_processDeletionsStreamSubscription = false,
		-- (Rx Subscription)

	_processCascadeDeletionsSubscription = false,
		-- (Rx Subscription)
		
	_continueFrom = false,
		-- (number)
		
	_previousContinueFrom = false,
		-- (number)

	WFCloseSupport.addCloseSupportMethods,

}

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

local getAlbumAssetIdsAffectedByDeletedAlbumSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )

	local albumAssetTable = "album_asset"

	if WFPlatformInfo.isDesktop then
		albumAssetTable = "album_asset_v2"
	end

	return string.format( "SELECT docId FROM %s WHERE albumId IN ( %s )",
			albumAssetTable,
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )

end )

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

local getAlbumAssetIdsAffectedByDeletedAssetSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	
	local albumAssetTable = "album_asset"

	if WFPlatformInfo.isDesktop then
		albumAssetTable = "album_asset_v2"
	end
	
	return string.format( "SELECT docId FROM %s WHERE assetId IN ( %s )",
			albumAssetTable,
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getCommentIdsAffectedByDeletedAssetSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
				
	return string.format( "SELECT docId FROM comment WHERE assetId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )

end )

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

local getVersionsIdsAffectedByDeletedAssetSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
				
	return string.format( "SELECT docId FROM versions WHERE assetId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )

end )

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

local getSpaceFavoriteIdsAffectedByDeletedAssetSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	

	return string.format( "SELECT docId FROM spacefavorite WHERE assetId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getCommentIdsAffectedByDeletedSpaceSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
				
	return string.format( "SELECT docId FROM comment WHERE spaceId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )

end )

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

local getVersionIdsAffectedByDeletedSpaceSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
				
	return string.format( "SELECT docId FROM versions WHERE spaceId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )

end )

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

local getSpaceFavoriteIdsAffectedByDeletedSpaceSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	

	return string.format( "SELECT docId FROM spacefavorite WHERE spaceId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getSpaceAlbumIdsAffectedByDeletedSpaceSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	

	return string.format( "SELECT docId FROM space_album WHERE spaceId IN ( %s )",
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getAssetIdsAffectedByDeletedSpaceFeedAlbumSQL 
		= KSFunctionUtils.memoizeFn( function( numberOfValues )
	
	-- The first SELECT gets all of the asset document IDs that exist within
	-- the deleted album(s).
	
	-- The second SELECT gets all of the asset document IDs that exist in other
	-- non-deleted albums.
	
	-- The EXCEPT of these is a list of asset documents that only exist
	-- within the deleted albums and therefore they need to be deleted.
	-- If we relied only on the first SELECT, we could delete assets that
	-- existed in other albums within the space.
	
	local albumAssetTableName = 'album_asset'
	
	if WFPlatformInfo.isDesktop then
		albumAssetTableName = 'album_asset_v2'	
	end
	
	return string.format( 
			[[SELECT DISTINCT assetId
				FROM %s
				WHERE albumId IN ( %s )
			
			EXCEPT

			SELECT DISTINCT assetId
				FROM %s
				 WHERE albumId NOT IN ( %s )]],
			albumAssetTableName,
			WFSQLiteUtils.createVariableArgList( numberOfValues ),
			albumAssetTableName,
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getSpaceAlbumIdsAffectedByDeletedAlbumSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	
	return string.format( 
			[[SELECT docId 
				FROM space_album
				WHERE albumId IN ( %s )]],
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local getSpaceIdsAffectedByDeletedAlbumSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	
	-- The following query gets the list of spaces that should be removed because the only
	-- album within the space was deleted.
	
	return string.format( 
			[[SELECT SA.spaceId
				FROM space_album SA
				WHERE SA.albumId IN ( %s )
					AND SA.spaceId NOT IN (
						SELECT DISTINCT space_album.spaceId
							FROM space_album
							WHERE space_album.albumId != SA.albumId )]],
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
		
end )

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

local function makeChunkedDeletionStream( makeGetDeletionIdsSQLFunc, docStore, indexStore, docIdsToDelete, startOffset, chunkSize )

	local idsToDeleteSQL = makeGetDeletionIdsSQLFunc( chunkSize )

	logger:infof( "Building deletion query: %s", idsToDeleteSQL )

	--[[ At this point idsToDeleteSQL is of the following form:
	
		SELECT ...
		FROM ...
		WHERE ... IN ( ?, ?, ... )
		
		Or of the form:
		
		SELECT ...
		FROM ...
		WHERE ... IN ( ?, ?, ... )
		...
		AND  ... IN ( ?, ?, ... )
		
		In the above SQL queries, a batch document IDs is passed into `queryOnce`
		in order to flesh out the "?" values. In the latter form, we are required to 
		pass in the documents IDs twice since they're referenced in two parts of the 
		query.
	]]--

	local batchOfDocIds = { unpack( docIdsToDelete, startOffset, startOffset + chunkSize - 1 ) }
	
	-- Here batchOfDocIds might look like this: { 1, 2, 3 }
	
	batchOfDocIds = KSTableUtils.appendArray( batchOfDocIds, { unpack( docIdsToDelete, startOffset, startOffset + chunkSize - 1 ) } ) 
		-- As noted above, some of the queries reference batchOfDocIds twice.
		-- See getAssetIdsAffectedByDeletedSpaceFeedAlbumSQL for an example of
		-- this. Queries that only refrence batchOfDocIds once will safely ignore
		-- the appended entries.
		
	-- Now batchOfDocIds might look like this: { 1, 2, 3, 1, 2, 3 }
		
	return indexStore:queryOnce {
			sql = idsToDeleteSQL,		 
			quiesceUntilUpToDate = true,
			singleColumnQuery = true,
			unpack( batchOfDocIds ),
		}
		
		:flatMapLatest( function( docIds )
		
			logger:infof( "Found %d documents to delete for query %s", #docIds, idsToDeleteSQL )
		
			return WFCatDelete.deleteDocs( docStore, docIds, true )
			
		end )		 

end

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

local function deleteRelatedDocumentsShared( docType, ... )

	local queries = {}

	if docType == 'asset' then
	
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getAlbumAssetIdsAffectedByDeletedAssetSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getCommentIdsAffectedByDeletedAssetSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getSpaceFavoriteIdsAffectedByDeletedAssetSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getVersionsIdsAffectedByDeletedAssetSQL, ... )
	
	elseif docType == 'album' then
	
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getAlbumAssetIdsAffectedByDeletedAlbumSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getSpaceAlbumIdsAffectedByDeletedAlbumSQL, ... )
	
	end
	
	return queries

end

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

local function deleteRelatedCatalogDocuments( docType, ... )

	local queries = {}

	if docType == 'space' then

		queries[ #queries + 1 ] = makeChunkedDeletionStream( getVersionIdsAffectedByDeletedSpaceSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getCommentIdsAffectedByDeletedSpaceSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getSpaceFavoriteIdsAffectedByDeletedSpaceSQL, ... )
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getSpaceAlbumIdsAffectedByDeletedSpaceSQL, ... )
			-- These cascade deletion queries only affect catalog documents. On the space-feed,
			-- a deleted space triggers the removal of all related documents. See processDeletedSpace.
		
	elseif docType == 'album' then
		
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getSpaceIdsAffectedByDeletedAlbumSQL, ... )
			-- This cascade deletion query only applies to catalog albums. When an album is deleted
			-- from the space-feed, it will create space tombstone on the feed-of-feeds if it was 
			-- the last album in the space.
		
	end
		
	KSTableUtils.appendArray( queries, deleteRelatedDocumentsShared( docType, ... ) )

	return queries

end

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

local function deleteRelatedSpaceFeedDocuments( docType, ... )

	local queries = {}

	-- If it's a deleted shared-with-you album, we must delete all associated 
	-- assets and album_assets. The asset must be removed first; if we delete
	-- the album_asset first, we don't know which assets need to be removed.
	
	if docType == 'album' then
	
		queries[ #queries + 1 ] = makeChunkedDeletionStream( getAssetIdsAffectedByDeletedSpaceFeedAlbumSQL, ... )

	end

	KSTableUtils.appendArray( queries, deleteRelatedDocumentsShared( docType, ... ) )

	return queries

end

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

local getVersionIdsAffectedByAssetTrashOperationSQL = KSFunctionUtils.memoizeFn( function( numberOfValues )
	
	return string.format( 
			[[SELECT docId, assetId
				FROM versions
				WHERE assetId IN ( %s )]],
			WFSQLiteUtils.createVariableArgList( numberOfValues ) )
					
end )

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

local function updateVersionsAffectedByAssetTrashOperation( docStore, indexStore, docIdsToUpdate, startOffset, chunkSize, newIdsByDocId )

	-- This function will cascade update the version documents associated
	-- with the trashed asset. This means that we will update the referenced
	-- asset Id within each version to instead reference the new asset (new_id).

	local idsToUpdateSQL = getVersionIdsAffectedByAssetTrashOperationSQL( chunkSize )

	logger:infof( "Building updateVersionsAffectedByAssetTrashOperation query: %s", idsToUpdateSQL )

	local batchOfDocIds = { unpack( docIdsToUpdate, startOffset, startOffset + chunkSize - 1 ) }
		
	return indexStore:queryOnce {
			sql = idsToUpdateSQL,		 
			quiesceUntilUpToDate = true,		 
			unpack( batchOfDocIds ),
		}
		
		:flatMapLatest( function( rows )
		
			if not rows or #rows < 1 then
				return Observable.just( true )
			end
			
			logger:infof( "Found %d versions to update", #rows )

			local overlays = {}
			overlays.label = '_dontsendtooz'
			
			for i, row in ipairs( rows ) do
			
				local versionId = row[ 1 ]
				local previousAssetId = row[ 2 ]
			
				overlays[ i ] = {
					_id = versionId,
					asset = {
						id = newIdsByDocId[ previousAssetId ]
					}
				}
					
			end

			return WFRxThrottledBatchedBulkApplyOverlay( "updateVersionsAffectedByAssetTrashOperation", docStore, overlays, 100, 10 )	
			
		end )		  			

end

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

local function updatedAssociatedDocumentsForTrashOperation( docType, ... )

	local queries = {}

	if docType == 'asset' then
	
		queries[ #queries + 1 ] = updateVersionsAffectedByAssetTrashOperation( ... )

	end

	return queries

end

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

local function cascadeDeleteDocuments( self, docType, docIds, deleteRelatedDocumentsFn, maybeNewIdsByDocId )

	local numberOfDocsToProcess = #docIds

	local offset = 1
	
	local queries = {}
	
	while offset <= numberOfDocsToProcess do
	
		local chunkSize = 0
		local remainingDocCount = numberOfDocsToProcess - offset
		
		if remainingDocCount >= 32 then
		
			chunkSize = 32
		
		elseif remainingDocCount >= 16 then
		
			chunkSize = 16
		
		elseif remainingDocCount >= 8 then
		
			chunkSize = 8
		
		elseif remainingDocCount >= 4 then
		
			chunkSize = 4
		
		elseif remainingDocCount >= 2 then
		
			chunkSize = 2
		
		else
		
			chunkSize = 1
		
		end
	
		logger:infof( "Now processing %d deleted docs of doctype %s", chunkSize, docType )
	
		local queriesToRun
				= deleteRelatedDocumentsFn( docType, self._docStore, self._indexStore, docIds, offset, chunkSize, maybeNewIdsByDocId )
	
		for _, query in ipairs( queriesToRun ) do
			queries[ #queries + 1 ] = query
		end
	
		offset = offset + chunkSize
	
	end

	return queries

end

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

local function processDeletedDocIds( self )

	local continueFrom = self._continueFrom
		-- It's possible for self._continueFrom to be modified after this function
		-- is called, but before the observables within this function complete. 
		-- We need to make sure use the continueFrom value that was present when
		-- this fn was called, hence the local.

	local cascadeDeletions = {}

	-- Build the cascade updates observables. 
	
	-- IMPORTANT: These trashed docs MUST process before the standard cascade deletions. 
	-- The reason for this is that the trashed docs may cause cascading updates to the 
	-- referenced ids within associated documents, which could then be affected by the 
	-- subsequent cascade deletions.
	--
	-- For example, if we received both a tombstone for when an asset was moved to the trash
	-- and also a tombstone for the asset when the trash was emptied, we would need to 
	-- process the trashed-tombstone first so that any document referencing the original asset
	-- is updated to reference the asset tombstones' new_id. Doing so would ensure that 
	-- when we process the trash-is-emptied tombstone we properly clean up all the associated
	-- docs.
	
	local trashedDocIdsToProcessByDocType = KSTableUtils.deepCopy( self._trashedDocIdsToProcessByDocType )
	self._trashedDocIdsToProcessByDocType = {}

	for docType, trashedDocsToProcess in pairs( trashedDocIdsToProcessByDocType ) do

		local docIdsToProcess = trashedDocsToProcess.docIds
		local newIdsByDocId = trashedDocsToProcess.newIdsByDocId

		local cascadeUpdatesForDocs = 
				cascadeDeleteDocuments( self, docType, docIdsToProcess, updatedAssociatedDocumentsForTrashOperation, newIdsByDocId )

		cascadeDeletions = KSTableUtils.appendArray( 
			cascadeDeletions, cascadeUpdatesForDocs )

	end

	-- Build the cascade deletion observables for the catalog-based deletions.
	
	local catalogDocIdsByDocType = KSTableUtils.deepCopy( self._catalogDocIdsToProcessByDocType )
	self._catalogDocIdsToProcessByDocType = {}
	
	for docType, catalogDocIds in pairs( catalogDocIdsByDocType ) do
	
		local cascadeDeletionsForCatalogDocs = 
				cascadeDeleteDocuments( self, docType, catalogDocIds, deleteRelatedCatalogDocuments )
	
		cascadeDeletions = KSTableUtils.appendArray( 
			cascadeDeletions, cascadeDeletionsForCatalogDocs )
	
	end
	
	-- Build the cascade deletion observables for the space-feed based deletions.
	
	local spaceFeedDocIdsByDocType = KSTableUtils.deepCopy( self._spaceFeedDocIdsToProcessByDocType )
	self._spaceFeedDocIdsToProcessByDocType = {}
	
	for docType, spaceFeedDocIds in pairs( spaceFeedDocIdsByDocType ) do
	
		local cascadeDeletionsForSpaceFeedDocs = 
				cascadeDeleteDocuments( self, docType, spaceFeedDocIds, deleteRelatedSpaceFeedDocuments )
	
		cascadeDeletions = KSTableUtils.appendArray( 
			cascadeDeletions, cascadeDeletionsForSpaceFeedDocs )
	
	end
	
	-- Now process the deletions

	if #cascadeDeletions > 0 then
		
		local disposeFn = function()
		
			self._processCascadeDeletionsSubscription:dispose()
			self._processCascadeDeletionsSubscription = nil
		
		end

		self._processCascadeDeletionsSubscription = Observable.from( cascadeDeletions )
		
			:map( function( cascadeDeletionQuery )

				return cascadeDeletionQuery
				
			end )
			
			:flatMerge( 1 )
				-- this prevents the queries from running concurrently.
		
			:subscribe( nil, 
				function( maybeErr )
				
					logger:errorf( "An error occured %s", maybeErr ) 
					disposeFn()
					
				end, 
				function()
				
					self._docStore:updateDeletedDocsSinceNamedChangeCounter( 
							kAutoDeleterChangeCounterLocalDocId, continueFrom )
						:call( runUntilDone )
					
					disposeFn()
					
				end )

	elseif type( continueFrom ) == 'number' and continueFrom > self._previousContinueFrom then

		-- Update the change counter here as well as items moved to trash
		-- or restored don't add anything to the cascadeDeletions.

		self._docStore:updateDeletedDocsSinceNamedChangeCounter(
				kAutoDeleterChangeCounterLocalDocId, continueFrom )
			:call( runUntilDone )

		self._previousContinueFrom = continueFrom

	end
	
end

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

local function processDeletedSpace( self, spaceId )

	WFCatSpacePurgeUtils.purgeAllForSpace( self._wfCatalog, spaceId, self._dataCache )
		:flatMap( function()
			return self._docStore:deleteSpace( spaceId )
		end )
		:call( runUntilDone )
	
end

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

function WFCatSyncAutoDeleter:__init( wfCatalog, optArgs, optDataCache )

	check( 'WFCatSyncAutoDeleter', 'WFCatalog', '?table', '?KSDataCache' )

	optArgs = optArgs or {}

	self._wfCatalog = wfCatalog
	self._dataCache = WFPlatformInfo.isAndroid and optDataCache or nil

	self._docStore = wfCatalog:docStore()
	self._docStore:addCloseTarget( self )

	self._indexStore = wfCatalog:indexStore()
	self._indexStore:addCloseTarget( self )

	self._catalogDocIdsToProcessByDocType = {}
	self._spaceFeedDocIdsToProcessByDocType = {}
	self._trashedDocIdsToProcessByDocType = {}
	
	self._previousContinueFrom = 0

	-- Queue up the deleted document IDs to process

	local docStore = self._docStore
	
	self._deletedDocIdStreamSubscription = docStore
			
		:deletedDocsSinceNamedChangeCounter( kAutoDeleterChangeCounterLocalDocId )

		:subscribe( function( docValues, continueFrom )
				
			for _, docValue in ipairs( docValues ) do
				
				local docType = docValue.type
				
				local new_id = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'new_id' )
						or docValue._newId
				
				local docId = docValue._id
				
				if docType == 'asset' and new_id then

					-- Trash assets have special handling within WFCatSyncAutoDeleter. In normal cases,
					-- trash operations should not trigger cascade deletions. However, if a second client
					-- issues a trash operation and then immediately empties the trash, then there's
					-- a chance Oz will not send the corresponding updates to the associated documents.
					-- This handling accounts for that.
					--
					-- This handling is currently only necessary for version documents. Most associated 
					-- documents (comments, favorites, album_assets, etc) are permanently deleted and
					-- recreated when the corresponding asset is trashed or restored; these tombstones
					-- are guarenteed to sync regardless of how quickly the trash is emptied.
					--
					-- See https://jira.corp.adobe.com/browse/WF-1235 for more info.

					local trashedDocsToProcess = self._trashedDocIdsToProcessByDocType[ docType ] or {}

					local trashedDocIds = trashedDocsToProcess.docIds or {}

					table.insert( trashedDocIds, docId )

					trashedDocsToProcess.docIds = trashedDocIds

					local newIdsByDocId = trashedDocsToProcess.newIdsByDocId or {}

					newIdsByDocId[ docId ] = new_id

					trashedDocsToProcess.newIdsByDocId = newIdsByDocId

					self._trashedDocIdsToProcessByDocType[ docType ] = trashedDocsToProcess

				elseif docType then
					
					if docValue._spaceId then
					
						local docsIdsToProcess = self._spaceFeedDocIdsToProcessByDocType[ docType ] or {}
						table.insert( docsIdsToProcess, docId )
						
						self._spaceFeedDocIdsToProcessByDocType[ docType ] = docsIdsToProcess
						
					else
					
						local docsIdsToProcess = self._catalogDocIdsToProcessByDocType[ docType ] or {}
						table.insert( docsIdsToProcess, docId )
						
						self._catalogDocIdsToProcessByDocType[ docType ] = docsIdsToProcess
						
					end
					
					if docType == 'space' then
					
						processDeletedSpace( self, docId )
						
					end
					
				end
			
			end
			
			self._continueFrom = continueFrom
			
		end )
		
	-- Every 15 seconds, process the deleted document IDs.

	self._processDeletionsStreamSubscription = Observable.timer( 0, optArgs.delay or 15, WFRxScheduler.sharedScheduler() )
		
		:filter( function()
		
			return not self._processCascadeDeletionsSubscription
				-- If we're already processing deleted doc ids, wait.
				
		end )
		
		:subscribe( function()
		
			processDeletedDocIds( self )
			
		end )
		
end

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

function WFCatSyncAutoDeleter:close()

	check( 'WFCatSyncAutoDeleter' )
	
	local deletedDocIdStreamSubscription = self._deletedDocIdStreamSubscription
	self._deletedDocIdStreamSubscription = nil
	
	if deletedDocIdStreamSubscription then
		deletedDocIdStreamSubscription:dispose()
	end

	local processDeletionsStreamSubscription = self._processDeletionsStreamSubscription
	self._processDeletionsStreamSubscription = nil
	
	if processDeletionsStreamSubscription then
		processDeletionsStreamSubscription:dispose()
	end
	
	local processCascadeDeletionsSubscription = self._processCascadeDeletionsSubscription
	self._processCascadeDeletionsSubscription = nil
	
	if processCascadeDeletionsSubscription then
		processCascadeDeletionsSubscription:dispose()
	end

end

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

return Rx.Class 'WFCatSyncAutoDeleter'( WFCatSyncAutoDeleter )
