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

WFCatSpaceFeedsManager.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2019 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.

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

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

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

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSJsonDecoder = require 'KSJsonDecoder'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFCloseSupport = require 'WFCloseSupport'
local WFOzDocUtils = require 'WFOzDocUtils'

	-- WF.core.rxlua
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxRetryObservable = require 'WFRxRetryObservable'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxThrottledBatchedApplyOverlayToIds = require 'WFRxThrottledBatchedApplyOverlayToIds'

	-- WF.oz_client.managed_catalog
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatDeniedListHelper = require 'WFCatDeniedListHelper'
local WFCatMetadataPuller = require 'WFCatMetadataPuller'
local WFCatMetadataPusher = require 'WFCatMetadataPusher'
local WFCatUrlUtils = require 'WFCatUrlUtils'
local WFCatPurgeHelper = require 'WFCatPurgeHelper'
local WFCatSpacePurgeUtils = require 'WFCatSpacePurgeUtils'

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

local WFCatSyncStatus

if WFPlatformInfo.isNimbus() then
	WFCatSyncStatus = require 'WFCatSyncStatus'
end

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

local logger = WFLogger( 'WFCatSpaceFeedsManager' )

local check = assert( WFArguments.check )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

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

local kEditor = 'editor'
local kContributor = 'contributor'
local kViewer = 'viewer'
local kAbandoned = 'abandoned'
local kAccepted = 'accepted'
	-- http://apidocs.adbecrsl.com/v2/data-model/space_user/index.html

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

local numberOfTriesBeforeDenyingSpace = 2
	-- Number of attempts to download a space_user doc from a space before 
	-- denying/blocking the space.

local deniedSpaceTimeoutInSeconds = 600
	-- Number of seconds a space remains blocked after we encounter unexpected sync
	-- errors.

local spacePullerMaxRetryAttempts = 3
	-- Number of attempts to fetch updates from Oz for a specific space before blocking
	-- the space.

local spacePullerRetryDelay = 60
	-- Number of seconds to wait in between retry attempts to fetch updates from a space.

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

--[[

Please see the following link for more info: 
https://git.corp.adobe.com/pages/oz/sync_docs/sync_feed

WFCatSpaceFeedsManager is responsible for all metadata sync activity related
to all shared-with-you spaces.

For syncing a user's standard catalog, there exists a single metadata puller 
(WFCatMetadataPuller) and a single metadata pusher (WFCatMetadataPusher). 
For syncing shared spaces, it's a bit different.

Here's how WFCatSpaceFeedsManager works:

1. There exists a metadata puller that fetches all the shared-with-you space 
documents using the My Spaces Feed (also called the feed-of-feeds). Similar 
to the catalog puller, the My Spaces Feed puller continuously polls Oz for 
updates. The documents in this feed also update whenever there are new updates
for each space; this data helps us know when to sync various space feeds.

2. For each active space we fetch via the My Spaces Feed, we create a new 
space-feed metadata puller for that space. These space-feed pullers DO NOT POLL
oz for updates. Instead, we use the information embedded in the space docs
fetched via the My Spaces Feed to know when there are new updates for a
particular space.

3. There exists a single instance of a metadata pusher that is used to push
all metadata updates, for all spaces, to Oz.

]]--

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

local WFCatSpaceFeedsManager = {

	_catalog = false,
		-- (WFCatalog)
	
	_mySpacesPuller = false,
		-- (WFCatMetadataPuller) A metadata puller used for the mySpaces feed.
	
	_spacePusher = false,
		-- (WFCatMetadataPusher) A single metadata pusher for all shared spaces.
	
	_latestSpacePusherActivityTimestampStream = false,
		-- (Rx.ReplaySubject) A stream of time in ISO8601 form which is the latest time space has activity,
		-- such as start continuous task, is doing work, or just finished pushing.
		-- We should delay shutdown space puller if we just notice space pusher activity
	
	_activeSpacesStream = false,
		-- (Rx.ReplaySubject) A stream that fires with information about active
		-- shared-with-you spaces.
		
	_inactiveSpacesStream = false,
		-- (Rx.ReplaySubject) A stream that fires with information about deleted
		-- and inactive shared-with-you spaces.
		
	_isDownloadingMetadataStream = false,
		-- (Rx.ReplaySubject) A stream that indicates if any space puller is 
		-- downloading metadata from Oz.
		
	_spacePullers = false,
	 	-- (table) A table of metadata pullers, one for each shared space.
		-- key: spaceUser ID
		-- value: WFCatMetadataPuller for the specific space. 
		
	_dataCache = false,
		-- (KSDataCache) KSDataCache database
		-- Only used by Android
		
	_activeSpaceHandlerSubscription = false,
 		-- (Rx Subscription)
		
	_inactiveSpaceHandlerSubscription = false,
		-- (Rx Subscription)
		
	_mySpacesUpdatesHandlerSubscription = false,
		-- (Rx Subscription)
		
	_isCatalogInitialSyncCompleteStream = false,
		-- (Rx.ReplaySubject)
		
	_canSendUpdatesStream = false,
		-- (Rx.ReplaySubject)
		
	_canFetchUpdatesStream = false,
		-- (Rx.ReplaySubject)
		
	_testOnly_canFetchUpdatesOnce = false,
		-- (Rx.ReplaySubject)
		
	_syncActivationSubscription = false,
		-- (Rx Subscription)
	
	_checkOzForSpaceUpdatesSubscription = false,
		-- (Rx Subscription)
		
	_checkForSpacePusherActivitySubscription = false,
		-- (Rx Subscription)
	
	_onDeletedSubscriptions = false,
		-- (table) A table of Rx subscriptions that monitor the pullers for deleted spaces.
		
	_spaceUserChangeSubscriptions = false,
		-- (table) A table of Rx subscriptions that monitor the space users for status and
		-- role changes.
		
	_spaceDeniedListHelper = false,
		-- (WFCatDeniedListHelper) A helper classed used to temporarily block spaces from syncing.
		-- These spaces can be blocked because they do not contain a space_user doc
		-- for this account. This can occur when Oz fails to delete a space once all 
		-- albums have been removed.
		
	_basicDocsForSpaceAreAvailableLocally = false,
		-- (table) A table used to keep track of which spaces have all the necessary
		-- docs local to display the album name in the UI.
		-- key: spaceUser ID
		-- value: boolean. True if the space has all necessary docs local.
		
	_spaceUserRoleCache = false,
		-- (table) A table used to keep track of the current users role for a specifed
		-- space.
		-- key: spaceUser ID
		-- value: the user's role. Either 'viewer' or 'contributor'
		
	_userEntitlementStatusCache = false,
		-- (table) WFCatUser entitlement.status string per spaceUser. To make sure all space 
		-- update entitlement correctly
		-- key: spaceUser ID
		-- value: entitlement.status string, will be set to 'invalid' if entitlement is nil
		
	_outOfQuotaBySpaceUserId = false,
		-- (table) A table used to track the space owner's quota status.
		-- key: spaceUserId
		-- value: (boolean) True if the owner is out of space. False otherwise. 
		
	_resetQuotaExceededTaskSubscriptions = false,
		-- (table) A table of Rx subscriptions that reset out-of-quota errors as needed.
		-- key: spaceUserId
		
	_entitlementExpiredBySpaceUserId = false,
		-- (table) A table used to track the space owner's entitlement status.
		-- key: spaceUserId
		-- value: (boolean) True if the owner's account is expired. False otherwise.
		
	_resetEntitlementExpiredTaskSubscriptions = false,
		-- (table) A table of Rx subscriptions that reset entitlement expired errors as needed.
		-- key: spaceUserId
		
	_spaceFeedsMaintenanceNotificationStream = false,
		-- (Rx ReplaySubject) Fires with updates when a shared-with-you space's availability
		-- changes due to Oz maintenance on the owner's account.
		
	_testMode = false,
		-- (boolean) DEBUG only; used in unit tests.

	_prioritizedSpaceId = false,
		-- (string) The space ID currently being prioritized for sync

	_prioritizedSpaceIdStream = false,
		-- (Rx ReplaySubject) Fires when _prioritizedSpaceId is updated.

	WFCloseSupport.addCloseSupportMethods,

}

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

-- TODO [jandrew 2019-08-01]: the version of remapOzDocument and convertUpdatesToDocValues
-- within this file are very similar to WFCatMetadataPuller's versions. At some point
-- we should consolidate them.

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

		ozDocument.deleted = nil
		result._deleted = true

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

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

		result._revIds = { KSIdGenerator.uuidInLowercaseHex(), force = true }

	end
	
	if spaceId then
		result._spaceId = spaceId
	end

	result._labels = {
		oz = true
	}

	return result
	
end

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

-- Fetch all space_user documents associated with the space via an Oz REST API.

local function getDocFromOzAndStoreLocally( self, spaceId, baseUrl, href )
	
	local catalogSpec = self._catalog:ozCatalogSpec()
	
	local options = {
		bypassRateLimiter = true,
	}
	
	local docUrl = WFCatUrlUtils.resolveToAbsoluteUrl( baseUrl, href )
			
	return catalogSpec:get( docUrl, options )
		
		:map( function( json, headers )
			
			if not headers or headers.status_code ~= '200' then
				error( "Request failed" )
			end
			
			local ozDoc = KSJsonDecoder.decodeWebSafeJson_stripNull( json )
			
			local docValue = remapOzDocument( ozDoc, baseUrl, spaceId )

			return docValue
			
		end )
		
		:flatMapLatest( function( docValue )
						
			return self._catalog:docStore():setDocValue( docValue )
			
				:map( function()

					return docValue
					
				end )
			
		end )
		
end

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

-- This function guarantees that all documents are present on disk that are needed
-- to display the minimal UI associated with a shared album. These documents include:
--
-- **album** document(s)
-- **space_album** document(s)
-- **feedbackuser** document for account owner
-- **space_user** document for the currently syncing user (see ensureSpaceUserDocIsAvailableLocally)
-- **feedbackuser** document for the currently syncing user

local function ensureBasicDocsAreAvailableLocally( self, spaceId )

	local docStore = self._catalog:docStore()

	return docStore:docValue( spaceId )

		:flatMap( function( spaceDoc )
			
			-- Gather all the IDs and hrefs for the documents that need to exist locally.
			
			local docIdsAndHrefs = {}
			local docsToFetch = {}
			
			local function storeIdAndHref( id, href )
			
				if id and href then
				
					docIdsAndHrefs[ id ] = href
					docsToFetch[ #docsToFetch + 1 ] = id
					
				end
				
			end
			
			local baseUrl = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'base' )
			
			-- Owner
			
			local ownerId = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'owner', 'id' )
			local ownerHref = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'owner', 'links',  'self', 'href' )
			
			storeIdAndHref( ownerId, ownerHref )
			
			-- Feedback User
			
			local feedbackUserId = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'this_space_user', 'user', 'id' )
			local feedbackUserHref = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'this_space_user', 'user', 'links',  'self', 'href' )
			
			storeIdAndHref( feedbackUserId, feedbackUserHref )
			
			-- Albums and Space Albums
			
			local albumsInSpaceDoc = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'albums' ) or {}
			
			for _, albumSection in ipairs( albumsInSpaceDoc ) do
			
				local albumId = KSTableUtils_derefOpt( albumSection, 'id' )
				local albumHref = KSTableUtils_derefOpt( albumSection, 'links',  'self', 'href' )
				
				storeIdAndHref( albumId, albumHref )
		
				local spaceAlbumId = KSTableUtils_derefOpt( albumSection, 'space_album', 'id' )
				local spaceAlbumHref = KSTableUtils_derefOpt( albumSection, 'space_album', 'links',  'self', 'href' )
				
				storeIdAndHref( spaceAlbumId, spaceAlbumHref )
				
			end
		
			-- Attempt to fetch all the necessary documents from the doc store.
			-- We'll need to download each document that does not exist locally.
			
			return docStore:bulkDocValue( docsToFetch )
			
				:map( function( docValues )
				
					return baseUrl, docValues, docIdsAndHrefs
		
				end )
		
		end )

		:flatMap( function( baseUrl, docValues, docIdsAndHrefs )
		
			local getDocsFromOzObservable = Observable.just()
		
			-- For each document that is empty (doesn't exist locally), fetch
			-- the doc from Oz and store it in the doc store.
		
			for _, docValue in ipairs( docValues ) do
			
				if docValue._empty then
				
					local href = docIdsAndHrefs[ docValue._id ]
					
					if href then
					
						getDocsFromOzObservable = getDocsFromOzObservable
								:concat( getDocFromOzAndStoreLocally( self, spaceId, baseUrl, href ) )
					
					end
					
				end	
			
			end
		
			return getDocsFromOzObservable
		
		end )
	
		:takeLast( 1 )
	
end

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

-- This function guarantees that the space_user document for this user and this 
-- space is in the local doc store. If it's not already present, it will fetch 
-- all space_user documents for the space from Oz and store them within the doc 
-- store.

local function ensureSpaceUserDocIsAvailableLocally( self, spaceId, spaceUserId )

	return self._catalog:docStore():docValue( spaceUserId )
	
		:flatMap( function( docValue )
			
			if docValue and not docValue._empty then
			
				return Observable.just( docValue )
					-- space_user doc is present.
					
			end
			
			return self._catalog:docStore():docValue( spaceId )
	
				:flatMap( function( spaceDoc )
				
					local baseUrl = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'base' )
				
					local spaceUserHref = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'this_space_user', 'links',  'self', 'href' )	
				
					return getDocFromOzAndStoreLocally( self, spaceId, baseUrl, spaceUserHref )
								
				end )
				
				:tapOnError( function( err )
					logger:errorf( "Adding spaceUser document failed for space: %s Error:%s", spaceId, err )
				end )
				
		end )
		
		:tapOnError( function( err )

			logger:errorf( "Adding spaceUser document failed. Error:%s", err )
			
		end )
		
end

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

local function intToBool( zeroOrOne )
	return zeroOrOne ~= 0
end

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

local function resetErrorsForDocIds( catalog, docIds )

	if #docIds == 0 then
		return Observable.just( true )
	end

	local overlay = WFCatOzErrors.createOverlayToRemoveOzError()

	return WFRxThrottledBatchedApplyOverlayToIds(
			"resetErrorsForDocIds", catalog:docStore(), overlay, docIds, 100, 10 )

end

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

local function resetErrorsForDocIdsStream( catalog, docIdsStream )

	return docIdsStream

		:flatMap( function( docIds )

			return resetErrorsForDocIds( catalog, docIds )

		end )

end

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

local function resetBinaryDownloadErrorsForDocIds( catalog, docIds )

	if #docIds == 0 then
		return Observable.just( true )
	end

	local overlay = {
		_localOnly = {
			binaryDownloadErrors = { _delete = true }
		}
	}

	return WFRxThrottledBatchedApplyOverlayToIds(
			"resetBinaryDownloadErrorsForDocIds", catalog:docStore(), overlay, docIds, 100, 10 )

end

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

local function resetQuotaExceededForSpaceTask( self, spaceId, spaceUserId )

	logger:infof( "resetQuotaExceededForSpaceTask called for spaceUserId %s", spaceUserId )
	
	local subscription = self._resetQuotaExceededTaskSubscriptions[ spaceUserId ] 

	if not subscription then
	
		logger:infof( "Creating a new resetQuotaExceededForSpaceTask subscription for spaceUserId %s", spaceUserId )
	
		local timerDelay = self._testMode and 0.1 or 60

		local docIdStream = Rx.Observable.timer( 0, timerDelay, WFRxScheduler.sharedScheduler() ) 

			:filter( function()
			
				return self._outOfQuotaBySpaceUserId[ spaceUserId ] == false
				
			end )

			:flatMap( function()
			
				return self._catalog:indexStore():queryOnce( 
					{ 
						sql = [[ SELECT docId 
									FROM quota_exceeded
									WHERE spaceId == ? ]],
						spaceId,
						quiesceUntilUpToDate = true,
						singleColumnQuery = true
					} )
			
			end )
			
			:tap( function( rows )
		
				logger:infof( "Now clearing quota expired errors for %d assets in space %s", #rows, spaceId )
				
			end )
			
		self._resetQuotaExceededTaskSubscriptions[ spaceUserId ] 
				= resetErrorsForDocIdsStream( self._catalog, docIdStream ):subscribe()

	end

end

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

local function resetEntitlementExpiredErrorsForSpaceTask( self, spaceId, spaceUserId )

	logger:infof( "resetEntitlementExpiredErrorsForSpaceTask called for spaceUserId %s", spaceUserId )

	local subscription = self._resetEntitlementExpiredTaskSubscriptions[ spaceUserId ] 

	if not subscription then

	 	logger:infof( "Creating a new resetEntitlementExpiredErrorsForSpaceTask subscription for spaceUserId %s", spaceUserId )

	 	local timerDelay = self._testMode and 0.1 or 60

		local entitlementExpiredUpdaterObs = Rx.Observable.timer( 0, timerDelay, WFRxScheduler.sharedScheduler() )

			:flatMap( function()
			
				local ownerEntitlementExpired = self._entitlementExpiredBySpaceUserId[ spaceUserId ]

				if not ownerEntitlementExpired then

					-- Remove entitlement related errors that are preventing uploads
					-- to the space. These errors are added to assets when we attempt
					-- to upload something to a space whose owner has an expired
					-- account. One the owner's entitlement is valid, we can clear
					-- these errors, which will result in retrying those uploads.

					logger:infof( "Owner entitlement is valid for space %s", spaceId )

					return self._catalog:indexStore():queryOnce( 
						{ 
							sql = [[ SELECT docId
										FROM blocked_uploads
										WHERE spaceId == ?
											AND ( reason == ? OR reason == ? ) ]],
							quiesceUntilUpToDate = true,
							singleColumnQuery = true,
							spaceId,
							WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError,
							WFCatOzErrors.Errors.kEntitlementExpiredError,
						} )

						:flatMap( function( docIds )

							if #docIds == 0 then
								return Observable.just( true )
							end

							logger:infof( "Now clearing entitlement expired errors for %d assets in space %s", #docIds, spaceId )

							return resetErrorsForDocIds( self._catalog, docIds )

						end )

				else if WFPlatformInfo.isNimbus() then

						-- The following code is Nimbus-only, as only LrD utilizes the
						-- binaryDownloadErrors _localOnly section.

						-- WF-1973: Lr clients hide shared album contents from the
						-- UI if the owner of the space has an expired account. However,
						-- that notification isn't immediate. It can take minutes for
						-- a client to receive it after the owner's account status changes.
						-- In the interim, if you try to view the album, the client may
						-- make requests to download renditions/negatives from the space.
						-- These requests will fail, as the links associated with the space
						-- will return 403s. In Nimbus, repeatedly requesting negatives may
						-- result in binaries being permanently deny-listed by the download
						-- code. As a workaround, when we detect that the owner's entitlement
						-- has expired, we periodically clear all binary download errors
						-- associated with the space. We do this for two reasons:
						--
						-- 1. It will allow us to download the binaries if the owner's subscription is
						-- resumed.
						--
						-- 2. It prevents these errors from showing up in the diagnostic log. There's
						-- no good reason to show binary download errors for assets associated with
						-- a space owned by an expired user.

						return WFCatSyncStatus.getLocalIdsWithBinaryDownloadErrorsForSpace( self._catalog, spaceId )

							:flatMap( function( docIds )

								if #docIds == 0 then
									return Observable.just( true )
								end

								logger:infof( "Now clearing binary download errors for %d assets in space %s", #docIds, spaceId )

								return resetBinaryDownloadErrorsForDocIds( self._catalog, docIds )

							end )

					end

				end

			end )

		self._resetEntitlementExpiredTaskSubscriptions[ spaceUserId ] 
				= entitlementExpiredUpdaterObs:subscribe()

	end

end

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

local function resetInsufficientPermissionErrorsForSpaceOnce( catalog, spaceId )

	logger:infof( "resetInsufficientPermissionErrorsForSpaceOnce called for space %s", spaceId )
			
	local docIdStream = catalog:indexStore():queryOnce( 
		{ 
			sql = [[ SELECT docId
						FROM blocked_uploads
						WHERE spaceId == ?
							AND reason == ? ]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true,
			spaceId,
			WFCatOzErrors.Errors.kInsufficientPermissionError
		} )
		
		:tap( function( rows )
		
			logger:infof( "Now clearing insufficient permission errors for %d assets in space %s", #rows, spaceId )
			
		end )
		
	return resetErrorsForDocIdsStream( catalog, docIdStream )

end

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

-- Monitor the index store for updates to space documents that are fetched
-- via the feed-of-feeds. These updates drive the spin-up/tear-down and sync
-- activity of the individual space pullers that are used to fetch updates
-- from single spaces.

local function monitorMySpacesUpdates( self )

	logger:info( "Monitoring all shared-with-you spaces for updates." )

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

	local migratedDeprecatedDocsForSpaceUserId = {}

	-- SPAM [jandrew 2019-09-05]: Using the index store to drive this logic
	-- may result in a slight delay. A better approach might be to use the
	-- doc store's `scannedFilteredMappedChangesFeedSince` API.
	
	return indexStore:query( 
		{ 
			sql = [[SELECT docId, spaceUserId, outOfQuota, isDeleted, 
							startingLink, currentUpdateLink, entitlementExpired,
							isUnderMaintenance
						FROM space]],
			quiesceUntilUpToDate = true,
			calm = 1
		} )

		:flatMap( function( rows )
		
			local spaces = {}
			
			for _, row in ipairs( rows ) do
			
				local space = {
					id = row[ 1 ],
					spaceUserId = row[ 2 ],
					isOutOfQuota = intToBool( row[ 3 ] ),
					isDeleted = intToBool( row[ 4 ] ),
					startingLink = row[ 5 ],
					currentUpdateLink = row[ 6 ],
					entitlementExpired = intToBool( row[ 7 ] ),
					isUnderMaintenance = intToBool( row[ 8 ] )
				}
				
				spaces[ #spaces + 1 ] = space
			
			end
		
			return Rx.Observable.from( spaces )
			
		end )

		:flatMap( function( space )
		
			local spaceId = space.id
			local spaceUserId = space.spaceUserId
			local entitlementExpired = space.entitlementExpired
			local isUnderMaintenance = space.isUnderMaintenance
			
			self._spaceFeedsMaintenanceNotificationStream:onNext( spaceId, isUnderMaintenance )

			if space.isDeleted and spaceUserId then
				self._inactiveSpacesStream:onNext( spaceUserId, false )
			end
			
			if ( entitlementExpired or isUnderMaintenance ) and spaceUserId then
				self._inactiveSpacesStream:onNext( spaceUserId, true )
			end
			
			if space.isDeleted then
				return Observable.just( false )
			end
			
			if not spaceId or not spaceUserId then
				return Observable.just( false )
			end
		
			if self._spaceDeniedListHelper:isDenied( spaceId ) then
				return Observable.just( false )
			end
			
			self._entitlementExpiredBySpaceUserId[ spaceUserId ] = not not entitlementExpired
			self._outOfQuotaBySpaceUserId[ spaceUserId ] = not not space.isOutOfQuota
			
			-- SPAM [jandrew 2019-10-02]: We used to use the space ID to form the unique
			-- ID used to track the next feed link for each space. Unfortunately this caused
			-- an issue if a user left a space, then rejoined the space. When this happened
			-- we deleted all local docs associated with the space (as expected), but those docs
			-- did not resync once the user rejoined the space. This was because the local doc we
			-- used to track the space's next link was/is not deleted when you leave a space.
			-- Essentially we resumed syncing where the user left off which prevented them from
			-- redownloading the local docs we removed. 
			-- NOTE: THIS ONLY AFFECTED QE AND INTERAL FOLKS.
			
			-- Deprecated Doc ID
			
			local deprecatedIdentifier = string.format( "Space %s Puller", spaceId )
			
			local deprecatedNextFeedLocalDocId = 
					WFCatMetadataPuller.getNextUpdateFeedLinkLocalDocId( false, deprecatedIdentifier )
			
			local deprecatedIsDoingInitialSyncLocalDocId = 
					WFCatMetadataPuller.getIsDoingInitialSyncLocalDocId( false, deprecatedIdentifier )
			
			-- Correct Doc ID
			
			local correctPullerIdentifier = string.format( "Space %s Puller", spaceUserId )
				-- Every time a user is added back to a space they are assigned a new spaceUserId.
				
			local spacePullerNextFeedLocalDocId = 
					WFCatMetadataPuller.getNextUpdateFeedLinkLocalDocId( false, correctPullerIdentifier )
					
			local spacePullerIsDoingInitialSyncLocalDocId = 
					WFCatMetadataPuller.getIsDoingInitialSyncLocalDocId( false, correctPullerIdentifier )		
					
			-- Deprecated Doc Migration.
			-- FIXME [jandrew 2019-10-03]: This migration logic is only temporary and should only affect
			-- internal QE and developers. This logic will be removed in a future release. 
			
			local maybeMigrateDeprecatedDocs = Observable.just()
			
			if not migratedDeprecatedDocsForSpaceUserId[ spaceUserId ] then
			
				-- If a user was previously using the deprecated local doc ID, then transfer the
				-- contents from the deprecated ID documents into new documents tracked by the
				-- correct IDs.
			
				maybeMigrateDeprecatedDocs = Observable.combineLatest(
						docStore:localDocValue( deprecatedNextFeedLocalDocId ),
						docStore:localDocValue( spacePullerNextFeedLocalDocId ),
						docStore:localDocValue( deprecatedIsDoingInitialSyncLocalDocId ) )
			
					:flatMap( function( nextFeedLocalDoc, realNextFeedLocalDoc, initialSyncLocalDoc )
						
						-- Only transfer the doc contents if there is no doc w/ the correct
						-- doc ID that already exists.
						
						if realNextFeedLocalDoc._empty and not nextFeedLocalDoc._empty then
						
							-- Move the docs from the deprecated IDs to the correct IDs.
							
							nextFeedLocalDoc._id = spacePullerNextFeedLocalDocId
							
							return docStore:setLocalDocValue( nextFeedLocalDoc )

								:flatMap( function()
								
									if not initialSyncLocalDoc._empty then
									
										initialSyncLocalDoc._id = spacePullerIsDoingInitialSyncLocalDocId
										return docStore:setLocalDocValue( initialSyncLocalDoc )
										
									end
								
									return Observable.just( true )
								
								end )

								:takeLast( 1 )
								
								:map( function()
									
									migratedDeprecatedDocsForSpaceUserId[ spaceUserId ] = true
									
									return true
								
								end )
						
						end
					
						migratedDeprecatedDocsForSpaceUserId[ spaceUserId ] = true
					
						return Observable.just( true )

					end )
		
			end
			
			return maybeMigrateDeprecatedDocs
			
				:flatMap( function()
			
					-- As soon as we see a space, we sync down the space_user documents associated
					-- with it before activating the metadata puller. This should guarantee the
					-- space_user doc is on disk before any asset document. The space document is
					-- required to know whether or not we should even sync the space.
					
					return ensureSpaceUserDocIsAvailableLocally( self, spaceId, spaceUserId )
					
				end )
		
				:flatMap( function( spaceUserDoc )
		
					assert( spaceUserDoc , string.format( "There is no space_user document on Oz for user %s. Blocking space %s from syncing.", spaceUserId, spaceId ) )
		
					local spaceUserStatus = KSTableUtils_derefOpt( spaceUserDoc, 'status' )
		
					local shouldSyncSpace = spaceUserStatus == kAccepted 
							and not entitlementExpired
							and not isUnderMaintenance
							
					local maybeFetchMoreDocuments = Observable.just()
					
					if not self._basicDocsForSpaceAreAvailableLocally[ spaceUserId ] then
					
						maybeFetchMoreDocuments = ensureBasicDocsAreAvailableLocally( self, spaceId ) 
					
					end

					return maybeFetchMoreDocuments
					
						:map( function()
						
							self._basicDocsForSpaceAreAvailableLocally[ spaceUserId ] = true
						
							-- If the space has expired, prevent us from sending any new updates.

							self._spacePusher:setSpaceIsExpired( spaceId, entitlementExpired )
						
							return spaceId, spaceUserId, shouldSyncSpace, space.startingLink, 
									space.isOutOfQuota, space.currentUpdateLink, correctPullerIdentifier
									
						end )
				
				end )
				
				:catch( function()
		
					self._spaceDeniedListHelper:maybeDeny( spaceId )
					return Observable.just( false )
				
				end )
		
		end )
		
		:subscribe( function( spaceId, spaceUserId, shouldSyncSpace, startingLink, isOutOfQuota, 
								currentUpdateLink, pullerIdentifier )
		
			if spaceId then
		
				if shouldSyncSpace then
				
					self._activeSpacesStream:onNext( spaceId, spaceUserId, startingLink, isOutOfQuota, 
														currentUpdateLink, pullerIdentifier  )
			
				end
			
			end
		
		end )
	
end

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

-- Setup a single instance of WFCatMetadataPusher that is used to push all
-- documents related to any of the spaces.

local function setupSpacePusher( self, optArgs )
	
	local spacePusherOptions = {
		delayBeforeSendingUpdates = 2,
		documentBatchCount = optArgs.pusherDocumentBatchCount or 100,
		albumAssetsForFilteredAssetsQueryFn = optArgs.albumAssetsForFilteredAssetsQueryFn,
		isSpacePusher = true,
		testMode = not not self._testMode, -- Only used by unit tests.
		capabilities = optArgs.capabilities, -- Only used by unit tests.
	}
	
	local spacePusher = WFCatMetadataPusher( self._catalog, spacePusherOptions )

	self:setOwned {
		_spacePusher = spacePusher,
	}
	
	-- Update latestSpacePusherActivityTimestamp when we see pusher is pushing updates
	
	self._checkForSpacePusherActivitySubscription = self._spacePusher:isDoingWork()
	
		:subscribe( function( isDoingWork )
	
			if isDoingWork then
				self._latestSpacePusherActivityTimestampStream:onNext( KSDate.timeInIso8601Format() )
			end
			
		end )
	
end

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

-- For a given space, shutdown the associated metadata puller. Clean up any
-- related handlers.

local function shutdownPullerForSpace( self, spaceUserId, shutdownShouldDelayWithFinalUpdates )

	local function shutdownPullerAndHandlers()
		
		-- Tear down the puller if it exists.
						
		local spacePuller = self._spacePullers[ spaceUserId ]
		self._spacePullers[ spaceUserId ] = nil
		
		if spacePuller then
			spacePuller:close()
		end

		-- Remove the on-deletion/space-user handlers.
		
		local subscription = self._onDeletedSubscriptions[ spaceUserId ]
		self._onDeletedSubscriptions[ spaceUserId ] = nil

		if subscription then
			subscription:dispose()
		end

		subscription = self._spaceUserChangeSubscriptions[ spaceUserId ]
		self._spaceUserChangeSubscriptions[ spaceUserId ] = nil

		if subscription then
			subscription:dispose()
		end
		
		-- Remove quota/expired account handlers.
		
		subscription = self._resetQuotaExceededTaskSubscriptions[ spaceUserId ]
		self._resetQuotaExceededTaskSubscriptions[ spaceUserId ] = nil

		if subscription then
			subscription:dispose()
		end
		
		subscription = self._resetEntitlementExpiredTaskSubscriptions[ spaceUserId ]
		self._resetEntitlementExpiredTaskSubscriptions[ spaceUserId ] = nil

		if subscription then
			subscription:dispose()
		end
		
	end

	if spaceUserId then
		
		if not shutdownShouldDelayWithFinalUpdates then
		
			shutdownPullerAndHandlers()
		
		else
		
			self._latestSpacePusherActivityTimestampStream
				
				:take( 1 )
			
				:flatMapLatest( function( spaceMightLatestPushTime )
				
					local now = KSDate.timeInIso8601Format()
					local timeSinceSpaceMightLatestPush = math.floor( KSDate.differenceInSeconds( spaceMightLatestPushTime, now ) )
					
					if timeSinceSpaceMightLatestPush >= 5 then
						return Observable.just( true )
					end
					
					local shutdownInSeconds = 5 - timeSinceSpaceMightLatestPush
					
					return Rx.Observable.timer( shutdownInSeconds, WFRxScheduler.sharedScheduler() )
				
				end )
				
				:flatMap( function()
				
					local spacePuller = self._spacePullers[ spaceUserId ]
					
					if spacePuller then
					
						return spacePuller:getLatestUpdates()
							
							:catch( function()
							
								return Observable.just( true )
								
							end )
							
					end
					
					return Observable.just( true )
					
				end )
				
				:tapOnCompleted( function()
		
					return shutdownPullerAndHandlers()
			
				end )
				
				:call( WFRxRunUntilDone )
				
		end
	end

end

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

-- Monitor Oz for evidence that a space has been deleted. isOzCatalogDeleted will
-- trigger when a 404 error is received with the "does not exist" message while 
-- attempting to fetch updates via the puller associated with the space.

local function monitorSpaceForDeletionViaSpaceFeed( self, spacePuller, spaceUserId )
	
	local subscription = self._onDeletedSubscriptions[ spaceUserId ]
	
	if not subscription then
	
		subscription = spacePuller:isOzCatalogDeleted()
				
			:filter( function( isDeleted )
				return isDeleted
			end )
			
			:take( 1 )
			
			:subscribe( function()
			
				self._inactiveSpacesStream:onNext( spaceUserId, false )
					
			end )
	
	end
	
	self._onDeletedSubscriptions[ spaceUserId ] = subscription
	
end

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

-- Monitor the spaceUser document to see if the status has changed to "abandoned", or
-- if the users role has changed.

-- If the status is "abandoned", then this means the user has left the space and we
-- need to immeadiately stop syncing down documents associated with the space. 
-- All space documents are deleted.

-- If the role changed from "viewer" to "contributor" we need to remove certain
-- errors from documents that are blocked from uploading.
		
local function monitorSpaceUser( self, spaceId, spaceUserId )
	
	local docStore = self._catalog:docStore()
	local indexStore = self._catalog:indexStore()
	
	local subscription = self._spaceUserChangeSubscriptions[ spaceUserId ]
	
	if not subscription then
	
		subscription = Observable.combineLatest( 
			
				function( spaceUserDocValue, entitlement )
				
					local spaceUserStatus = KSTableUtils_derefOpt( spaceUserDocValue, 'status' )
					local spaceUserRole = KSTableUtils_derefOpt( spaceUserDocValue, 'role' )
					local entitlementStatus = KSTableUtils_derefOpt( entitlement, 'status' ) or 'invalid'
					local downgradedFromEditorRole = KSTableUtils_derefOpt( spaceUserDocValue, '_localOnly', 'universal', 'downgradedFromEditorRole' ) or false
					
					return spaceUserStatus, spaceUserRole, entitlementStatus, downgradedFromEditorRole
					
				end,
				docStore:ongoingDocValue( spaceUserId ),
				self._catalog:ozUser():entitlement()
			)
			
			:distinctUntilChanged()
			
			:flatMap( function( spaceUserStatus, spaceUserRole, entitlementStatus, downgradedFromEditorRole ) 
			
				local updateSpaceUserInfoCache = function( spaceUserRole, entitlementStatus )
				
					self._spaceUserRoleCache[ spaceUserId ] = spaceUserRole
					self._userEntitlementStatusCache[ spaceUserId ] = entitlementStatus
					
				end
			
				if spaceUserStatus == kAbandoned then
				
					updateSpaceUserInfoCache( spaceUserRole, entitlementStatus )
				
					-- Shutdown the metadata puller.
					
					shutdownPullerForSpace( self, spaceUserId, false )
				
					-- Here we issue a call to the doc store to delete the space. This will 
					-- remove all of the documents associated with the space from the doc store.
					-- We specifically exclude the deletion of the space_user document because
					-- we need the 'abandoned' status to sync to Oz. Once it has synced, a
					-- tombstone for the space will show up on the feed-of-feeds resulting in
					-- the cleanup of the space_user document.
					
					local excludedDocumentIds
					
					if spaceUserId then
						excludedDocumentIds = { spaceUserId }
					end
					
					return WFCatSpacePurgeUtils.purgeAllForSpace( self._catalog, spaceId, self._dataCache )
						
						:flatMap( function()
							
							return docStore:deleteSpace( spaceId, excludedDocumentIds )
						
						end )
				
				elseif ( ( self._spaceUserRoleCache[ spaceUserId ] == kEditor 
								or ( downgradedFromEditorRole and not self._spaceUserRoleCache[ spaceUserId ] ) )
							and ( spaceUserRole == kViewer or spaceUserRole == kContributor ) ) then
							
					-- Purge asset master/proxy and xmp_develop files
					-- Purge all version files
					
					-- Note that desktop xmp_develop files won't be purged here, it would be handled by PixelProviderFileCachePurgeWorker
					
					updateSpaceUserInfoCache( spaceUserRole, entitlementStatus )
					
					local spaceUserOverlay = {
						_id = spaceUserId,
						_localOnly = {
							universal = {
								downgradedFromEditorRole = true,
							}
						}
					}
					
					-- offline use flag should be removed from all albums in space
					
					local albumIdName = WFPlatformInfo.isDesktop and "albumId" or "docId"
					
					return indexStore:queryOnce {
							sql = "SELECT " .. albumIdName .. " FROM albums WHERE spaceId == ?", spaceId,
							quiesceUntilUpToDate = true,
							singleColumnQuery = true
						}

						:flatMap( function( albumIds )
						
							local overlays = {}
							local deletionModifier = { _delete = true }
							
							for _, albumId in pairs( albumIds ) do
								local albumOverlay = {
									_id = albumId,
									_localOnly = {
										isMarkedForOfflineUse = WFPlatformInfo.isDesktop and deletionModifier or nil,
										useOffline = WFPlatformInfo.isMobile() and deletionModifier or nil
									}
								}
								
								overlays[ #overlays + 1 ] = albumOverlay
							end
							
							overlays[ #overlays + 1 ] = spaceUserOverlay
							
							return docStore:bulkApplyOverlay( overlays )
						
						end )
						
						:flatMap( function()
						
							return WFCatSpacePurgeUtils.purgeEditForSpace( self._catalog, spaceId )
							
						end )
						
						:flatMap( function()
						
							return docStore:applyOverlay {
								_id = spaceUserId,
								_localOnly = {
									universal = {
										downgradedFromEditorRole = { _delete = true },
									}
								}
							}
						
						end )
				
				elseif (( not self._spaceUserRoleCache[ spaceUserId ] or self._spaceUserRoleCache[ spaceUserId ] == kViewer )
							and ( spaceUserRole == kContributor or spaceUserRole == kEditor ))
						or ( self._spaceUserRoleCache[ spaceUserId ] == kContributor
							and spaceUserRole == kEditor ) then
					
					updateSpaceUserInfoCache( spaceUserRole, entitlementStatus )
					
					return resetInsufficientPermissionErrorsForSpaceOnce( self._catalog, spaceId )

				elseif spaceUserRole == kEditor 
							and ( self._userEntitlementStatusCache[ spaceUserId ] == 'invalid'
									or self._userEntitlementStatusCache[ spaceUserId ] == 'trial_expired' 
									or self._userEntitlementStatusCache[ spaceUserId ] == 'subscription_expired' )
							and ( entitlementStatus == 'trial' 
									or entitlementStatus == 'subscription' ) then
					
					updateSpaceUserInfoCache( spaceUserRole, entitlementStatus )
					
					return resetInsufficientPermissionErrorsForSpaceOnce( self._catalog, spaceId )
				
				end
					
				updateSpaceUserInfoCache( spaceUserRole, entitlementStatus )
				
				return Observable.just( true )
					
			end )
			
			:catch( function( errorMessage )
			
				logger:warn( "An unknown error occured: %s", errorMessage )
				return Observable.just( true )
				
			end )
				
			:subscribe()
		
	end
	
	self._spaceUserChangeSubscriptions[ spaceUserId ] = subscription
	
end

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

local function setupSpacePullers( self, optArgs )

	-- Spin up the pullers / pusher as needed based on the responses
	-- from Oz.
	
	self._activeSpaceHandlerSubscription = self._activeSpacesStream
	
		:map( function( spaceId, spaceUserId, startingLink, isOutOfQuota, currentUpdateLink, pullerIdentifier )
		
			logger:infof( [[Active space stream update: 
					space ID: %s, startingLink: %s /n
					isOutOfQuota: %s, currentUpdateLink: %s]],
					spaceId, startingLink, isOutOfQuota, currentUpdateLink )

			-- Create a puller for the space if one doesn't already exist.
			
			local spacePuller = self._spacePullers[ spaceUserId ]
			
			if not spacePuller then
			
				logger:infof( "Spinning up a new WFCatMetadataPuller for spaceId %s", spaceId )
			
				local pullerOptions = {

					prefetch = optArgs.isPrefetchEnabled, 
					documentBatchCount = optArgs.pullerDocumentBatchCount,
					spaceId = spaceId, 
					updatesFeedUrl = startingLink,
					isCatalogPuller = false,
					capabilities = optArgs.capabilities, -- Only used by unit tests.

				}
			
				spacePuller = WFCatMetadataPuller( self._catalog, pullerIdentifier, pullerOptions )
					
				self._spacePullers[ spaceUserId ] = spacePuller
					
				monitorSpaceForDeletionViaSpaceFeed( self, spacePuller, spaceUserId )
				
				monitorSpaceUser( self, spaceId, spaceUserId )

				-- Setup ongoing queries that periodically attempt to reset docs in error
				-- states associated with the space.
				
				resetQuotaExceededForSpaceTask( self, spaceId, spaceUserId )

				resetEntitlementExpiredErrorsForSpaceTask( self, spaceId, spaceUserId )

			end
			
			self._spacePusher:setSpaceIsOverQuota( spaceId, isOutOfQuota )
			
			-- If there are space feed updates trigger the puller to fetch them.
			
			-- Unlike the catalog metadata puller, the space-based pullers
			-- do not periodically poll Oz for updates. Instead, we poll
			-- the Oz feed-of-feeds API for updates across all spaces. If we
			-- detect an update, we then trigger the appropriate metadata
			-- puller.

			-- We're wrapping this call in WFRxRetryObservable so that we will attempt
			-- to fetch updates again in the event of an intermittent Oz hiccup. 
			-- If we're unable to fetch updates after retrying a few times, we
			-- temporarily block the space; this is to prevent us from repeatedly
			-- issuing HTTP requests on a space we may have lost access to.
			
			local function retryIfNot4XXs( responseData, responseHeader )

				local statusCode = responseHeader and responseHeader.status_code or ''

				if string.sub( statusCode, 1, 1 ) == '4' then
				
					return false -- 4XX HTTP status code, so abort the retry.
					
				end

				return true -- try again

			end
			
			local retryDelay = self._testMode and 0.1 or spacePullerRetryDelay
			
			local result = WFRxRetryObservable 
					{
						maxAttempts = spacePullerMaxRetryAttempts,
						retryDelay = retryDelay,
						retryFilter = retryIfNot4XXs,
						factory = function()

							return Observable.combineLatest(
									self._canFetchUpdatesStream,
									self._testOnly_canFetchUpdatesOnce,
									self._prioritizedSpaceIdStream )

								:take( 1 )

								:filter( function( canFetchUpdates, testOnly_canFetchUpdatesOnce, maybePrioritizedSpaceId )

									local shouldFetchUpdates = ( canFetchUpdates
											and ( not maybePrioritizedSpaceId or maybePrioritizedSpaceId == spaceId ) )
											or testOnly_canFetchUpdatesOnce

									return shouldFetchUpdates

								end )

								:flatMap( function()
									return spacePuller:getLatestUpdatesIfNeeded( currentUpdateLink )
								end )

						end
					}
				
				:catch( function( err )
		
					-- Temporarily block the space.
					
					logger:warnf( "Failure to fetch space updates for spaceId %s: %s", spaceId, err )
					
					self._spaceDeniedListHelper:addToDeniedList( spaceId )
						
					return Observable.just()
					
				end )
				
			return result
			
		end )
		
		:flatMerge( self._testMode and 1 or 2 ) 
			-- By allowing two space-feeds to sync at a time, this prevents sync from feeling
			-- completely stalled in te the event one space is encountering Oz hiccups.
			-- In test mode, we limit concurrency to make test results more predictable.
		
		:subscribe()
	
	-- When we notice a space should be inactive, we shutdown the WFCatMetadataPuller
	-- associated with the space. _inactiveSpacesStream can be trigered by:
	--
	-- 1.) The space owner's account entitlement expires.
	--
	-- 2.) A 404 error received with the "does not exist" message when attempting to
	--		fetch updates via the puller.
	--
	-- 3.) The space owner's account is under maintenance.

	self._inactiveSpaceHandlerSubscription = self._inactiveSpacesStream
	
		:subscribe( function( spaceUserId, shutdownShouldDelayWithFinalUpdates )
		
			logger:infof( "Inactive space stream update: %s", spaceUserId )
		
			shutdownPullerForSpace( self, spaceUserId, shutdownShouldDelayWithFinalUpdates )

			-- Note that the deletion of the space-feed documents happens automatically
			-- when the space tombstone is delivered on the feed-of-feeds via the
			-- WFCatSyncAutoDeleter.

		end )

	-- Trigger the feed-of-feeds API to check for updates after we
	-- push changes to one of the spaces.
	
	self._checkOzForSpaceUpdatesSubscription = self._spacePusher:workHasCompletedStream()
	
		:subscribe( function()
		
			logger:info( "Space pusher work detected. Will soon check Oz for updates." )
			
			self._latestSpacePusherActivityTimestampStream:onNext( KSDate.timeInIso8601Format() )
			
			self._mySpacesPuller:checkOzForUpdates()
				
		end )

	-- Setup the main Oz feed-of-feeds puller
	
	local pullerOptions = {
	
		continuousTaskDelays = optArgs.taskDelays or 
			{
				progressiveBackoff = true,
				backoffStrategy = WFRxContinuousTask.BackoffStrategy_Exponential,
				noWorkDelay = 2,
				maxNoWorkDelay = 60,
				errorDelay = 10,
				maxDelay = 60,
			},
			
		updatesFeedUrl = self._catalog:ozCatalogSpec():spaceUploadUrl(),
		
		isCatalogPuller = false
		
	}
	
	local mySpacesPuller = WFCatMetadataPuller( self._catalog, "Feed-of-Feeds Puller", pullerOptions )
	
	self:setOwned {
		_mySpacesPuller = mySpacesPuller,
	}
	
	self._mySpacesUpdatesHandlerSubscription = monitorMySpacesUpdates( self )
	
end

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

local function maybeSetupIsDownloadingMetadataStream( self )

	if not self._isDownloadingMetadataStream then
	
		self._isDownloadingMetadataStream = Observable.combineLatest( 
                self._activeSpacesStream:startWith( true ), 
                self._inactiveSpacesStream:startWith( true ) )
		
			:map( function()
			
				-- If either _activeSpacesStream or _inactiveSpacesStream fire,
				-- we know the list of space pullers may have changed.
			
				return self._spacePullers
				
			end )
	
			:flatMapLatest( function( spacePullers )
	
				local pullerStatusStreams = {}
				
				for _, puller in pairs( spacePullers ) do
				
					pullerStatusStreams[ #pullerStatusStreams + 1 ] = puller:isDoingWork()
				
				end
				
				local result = Observable.just( false )
				
				if #pullerStatusStreams > 0 then
				
					result = Observable.combineLatest( unpack( pullerStatusStreams ) )
					
						:map( function( ... )
						
							for _, isPullingMetadata in ipairs( arg ) do

								if isPullingMetadata then
								
									return true
									
								end
								
							end
							
							return false
						
						end )
				
				end
				
				return result
				
			end )
			
			:distinctUntilChanged()
			
		if not self._testMode then
	
			self._isDownloadingMetadataStream = self._isDownloadingMetadataStream
			
				:calm( 3, WFRxScheduler.sharedScheduler() )
				
		end	

	end

end

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

local function startSendingUpdates( self )

	self._latestSpacePusherActivityTimestampStream:onNext( KSDate.timeInIso8601Format() )

	self._spacePusher:sendUpdatesContinuously()

end

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

local function stopSendingUpdates( self )

	self._spacePusher:stopContinuousUpdates()

end

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

local function startFetchingUpdates( self )

	self._mySpacesPuller:getUpdatesContinuously()

end

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

local function stopFetchingUpdates( self )

	self._mySpacesPuller:stopContinuousUpdates()

end

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

--@mark WFCatSpaceFeedsManager Class

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

--[[!
	\brief Initialize a new WFCatSpaceFeedsManager object.
	
	\param catalog (WFCatalog)
	\param optArgs (table): valid keys as follows:
		taskDelays (table)
		pullerDocumentBatchCount (number)
		pusherDocumentBatchCount (number)
		albumAssetsForFilteredAssetsQueryFn (function)
		testMode (boolean)
		
	\return WFCatSpaceFeedsManager
--]]

function WFCatSpaceFeedsManager:__init( catalog, optArgs )

	check( 'WFCatSpaceFeedsManager', 'WFCatalog', '?table' )

	self:setOwned {
	
		_spaceDeniedListHelper = WFCatDeniedListHelper( 
				catalog:docStore(), 
				'WFCatSpaceFeedsManager',
				numberOfTriesBeforeDenyingSpace,
				deniedSpaceTimeoutInSeconds,
				false ),
	
		_catalog = catalog:relayObject()
		
	}

	optArgs = optArgs or {}
	
	self._testMode = optArgs.testMode

	self._dataCache = WFPlatformInfo.isAndroid and optArgs.dataCache or nil
	
	self._spacePullers = {}

	self._onDeletedSubscriptions = {}

	self._spaceUserChangeSubscriptions = {}

	self._activeSpacesStream = Rx.ReplaySubject( 1 )
	self._inactiveSpacesStream = Rx.ReplaySubject( 1 )

	self._isCatalogInitialSyncCompleteStream = Rx.ReplaySubject( 1 )
	self._canSendUpdatesStream = Rx.ReplaySubject( 1 )
	self._canFetchUpdatesStream = Rx.ReplaySubject( 1 )

	self._prioritizedSpaceIdStream = Rx.ReplaySubject( 1 )
	self._prioritizedSpaceIdStream:onNext( false )

	self._spaceFeedsMaintenanceNotificationStream = Rx.ReplaySubject( 1 )
	
	self._testOnly_canFetchUpdatesOnce = Rx.ReplaySubject( 1 )
	self._testOnly_canFetchUpdatesOnce:onNext( false )
	
	self._basicDocsForSpaceAreAvailableLocally = {}
	
	self._spaceUserRoleCache = {}
	self._userEntitlementStatusCache = {}
	
	self._entitlementExpiredBySpaceUserId = {}
	self._resetEntitlementExpiredTaskSubscriptions = {}
	
	self._outOfQuotaBySpaceUserId = {}
	self._resetQuotaExceededTaskSubscriptions = {}
	
	self._latestSpacePusherActivityTimestampStream = Rx.ReplaySubject( 1 )
	self._latestSpacePusherActivityTimestampStream:onNext( KSDate.timeInIso8601Format() )
	
	setupSpacePusher( self, optArgs )
	
	-- Monitor the Oz feed-of-feeds API and setup space-based pullers as needed.
	
	setupSpacePullers( self, optArgs )
	
	-- WF-398: Prioritize sending/receiving catalog-feed updates over space-feed updates.
	
	local syncActivationStream = Observable.combineLatest(
			self._isCatalogInitialSyncCompleteStream:startWith( false ),
			self._canSendUpdatesStream:startWith( false ),
			self._canFetchUpdatesStream:startWith( false ),
			self._prioritizedSpaceIdStream:startWith( false ) )
		
	self._syncActivationSubscription = syncActivationStream
	
		:subscribe( function( isCatalogInitialSyncComplete, canSendUpdates, canFetchUpdates, prioritizedSpaceId )
		
			-- When prioritizing space sync or during initial sync, we have high priority sync activity
			local initialSyncTakesPriority = not isCatalogInitialSyncComplete and not prioritizedSpaceId
		
			if initialSyncTakesPriority then
				
				stopSendingUpdates( self )
				stopFetchingUpdates( self )
			
			else
				
				if canSendUpdates then
					startSendingUpdates( self )
				else
					stopSendingUpdates( self )
				end
				
				if canFetchUpdates then
					startFetchingUpdates( self )
				else
					stopFetchingUpdates( self )
				end
			
			end
	
		end )
		
	if self._testMode then
		WFCatPurgeHelper.DEBUG_optOutDeletionBacklog = true
	end
	
end

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

--[[!
	\brief Closes the object.

	\return void
--]]

function WFCatSpaceFeedsManager:close()
 
	check( 'WFCatSpaceFeedsManager' )

	local classDisposables = {
		'_activeSpacesStream',
		'_inactiveSpacesStream',
		'_activeSpaceHandlerSubscription',
		'_inactiveSpaceHandlerSubscription',
		'_checkOzForSpaceUpdatesSubscription',
		'_checkForSpacePusherActivitySubscription',
		'_isCatalogInitialSyncCompleteStream',
		'_canSendUpdatesStream',
		'_canFetchUpdatesStream',
		'_syncActivationSubscription',
		'_mySpacesUpdatesHandlerSubscription',
		'_spaceFeedsMaintenanceNotificationStream',
		'_prioritizedSpaceIdStream'
	}

	for _, key in ipairs( classDisposables ) do
	
		local disposable = self[ key ]
		self[ key ] = nil
		
		if disposable then
			disposable:dispose()
		end
		
	end

	local classTablesOfDisposables = {
		'_onDeletedSubscriptions',
		'_spaceUserChangeSubscriptions',
		'_resetEntitlementExpiredTaskSubscriptions',
		'_resetQuotaExceededTaskSubscriptions',
	}

	for _, key in ipairs( classTablesOfDisposables ) do
	
		for _, disposable in pairs( self[ key ] ) do
	
			if disposable then
				disposable:dispose()
			end
			
		end
		
		self[ key ] = {}
		
	end

end

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

--[[!
	\brief Enable the space metadata pusher.

	\return void
--]]

function WFCatSpaceFeedsManager:startSendingUpdates()

	check( 'WFCatSpaceFeedsManager' )

	self._canSendUpdatesStream:onNext( true )

end

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

--[[!
	\brief Disable the space metadata pusher.

	\return void
--]]

function WFCatSpaceFeedsManager:stopSendingUpdates()

	check( 'WFCatSpaceFeedsManager' )

	self._canSendUpdatesStream:onNext( false )

end

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

--[[!
	\brief Enable polling of the Oz feed-of-feeds API.

	\return void
--]]

function WFCatSpaceFeedsManager:startFetchingUpdates()

	check( 'WFCatSpaceFeedsManager' )

	self._canFetchUpdatesStream:onNext( true )

end

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

--[[!
	\brief Disable polling of the Oz feed-of-feeds API.

	\return void
--]]

function WFCatSpaceFeedsManager:stopFetchingUpdates()

	check( 'WFCatSpaceFeedsManager' )

	self._canFetchUpdatesStream:onNext( false )

end

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

--[[!
	\brief Indicates if the space pusher is uploading metadata to Oz.

	\return Observable(boolean) True if the space pusher is uploading
		metadata to Oz.
--]]

function WFCatSpaceFeedsManager:isUploadingMetadata()

	check( 'WFCatSpaceFeedsManager' )
	
	local result = Observable.just( false )
	
	result = self._spacePusher:isDoingWork():distinctUntilChanged()
	
	return result
	
end

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

--[[!
	\brief Indicates if any space puller is downloading metadata from Oz.

	\return Observable(boolean) True if a space puller is downloading
		metadata from Oz.
--]]

function WFCatSpaceFeedsManager:isDownloadingMetadata()

	check( 'WFCatSpaceFeedsManager' )
	
	maybeSetupIsDownloadingMetadataStream( self )
	
	return self._isDownloadingMetadataStream
	
end

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

--[[!
	\brief Let the space feed manager know that the catalog initial sync is complete.

	\return nil
--]]

function WFCatSpaceFeedsManager:reportCatalogInitialSyncIsComplete( optIsComplete )

	check( 'WFCatSpaceFeedsManager' )
	
	self._isCatalogInitialSyncCompleteStream:onNext( optIsComplete == nil or optIsComplete )
	
end

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

--[[!
	\brief Get a stream that fires with updates when a shared-with-you space's 
		availability changes due to Oz maintenance on the owner's account.

	\return Observable(spaceId, status) Status is 'true' in the event
		that the owner's account is under maintenance or down, otherwise
		it will be 'false'.
--]]

function WFCatSpaceFeedsManager:spaceFeedsMaintenanceNotificationStream()

	check( 'WFCatSpaceFeedsManager' )
	
	return self._spaceFeedsMaintenanceNotificationStream

		:map( function( spaceId, isUnderMaintenance )
		
			return {
				id = spaceId,
				isUnderMaintenance = isUnderMaintenance
			}
		
		end )
	
end

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

--[[!
	\brief This API that can be used gather real-time sync diagnostic data from 
		the space feeds manager. 

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

	\return Observable< table > An Observable whose value tracks a table of
		sync diagnostic data.
--]]

function WFCatSpaceFeedsManager:syncDiagnostics( optCalm )

	check( 'WFCatSpaceFeedsManager', '?number' )

	local spacePullersSyncDiagnosticStreams = {}
				
	for _, puller in pairs( self._spacePullers ) do
	
		spacePullersSyncDiagnosticStreams[ #spacePullersSyncDiagnosticStreams + 1 ] 
				= puller:syncDiagnostics( optCalm )
	
	end

	return Observable.combineLatest(
			self._spacePusher:syncDiagnostics( optCalm ),
			self._mySpacesPuller:syncDiagnostics( optCalm ),
			unpack( spacePullersSyncDiagnosticStreams ) )
			
		:map( function( spacePusherSyncDiagnostics, feedOfFeedsPullerSyncDiagnostic, ... )
		
			return {
				spacePusherSyncDiagnostics = spacePusherSyncDiagnostics,
				feedOfFeedsPullerSyncDiagnostics = feedOfFeedsPullerSyncDiagnostic,
				spacePullersSyncDiagnostics = { ... }
			}
		
		end )

end

--------------------------------------------------------------------------------
--@mark Test-Only Functions
--------------------------------------------------------------------------------

function WFCatSpaceFeedsManager:_testOnly_getUpdatesOnce()

	check( 'WFCatSpaceFeedsManager' )

	self._canFetchUpdatesStream:onNext( false )
	self._testOnly_canFetchUpdatesOnce:onNext( true )
	
	return self._mySpacesPuller:getUpdatesOnce( self )

end

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

function WFCatSpaceFeedsManager:_testOnly_canFetchUpdates()

	check( 'WFCatSpaceFeedsManager' )

	return self._mySpacesPuller:_testOnly_getIsTaskRunningStream()

end

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

function WFCatSpaceFeedsManager:_testOnly_canSendUpdates()

	check( 'WFCatSpaceFeedsManager' )

	return self._spacePusher:_testOnly_getIsTaskRunningStream()

end

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

function WFCatSpaceFeedsManager:_testOnly_resetEntitlementExpiredErrorsForSpaceTask( spaceId, spaceUserId )

	check( 'WFCatSpaceFeedsManager', '+string', '+string' )

	return resetEntitlementExpiredErrorsForSpaceTask( self, spaceId, spaceUserId )

end

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

--[[!
	\brief Prioritize sync for a specific space
	
	This method activates space prioritization mode, during which normal sync 
	activation logic is bypassed and the specified space is synced using its 
	space puller. The prioritization will be deactivated by the caller.
	
	\param spaceId The ID of the space to prioritize for sync
	
	\return boolean True if prioritization was successfully activated
--]]

function WFCatSpaceFeedsManager:prioritizeSpaceSync( spaceId )

	check( 'WFCatSpaceFeedsManager', '+string' )

	logger:infof( "Activating space sync prioritization for space %s", spaceId )

	self._prioritizedSpaceId = spaceId

	self._prioritizedSpaceIdStream:onNext( spaceId )

end

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

--[[!
	\brief Stop space sync prioritization
	
	This method deactivates space prioritization mode, allowing normal sync 
	activation logic to resume.
--]]

function WFCatSpaceFeedsManager:stopPrioritizeSpaceSync()

	check( 'WFCatSpaceFeedsManager' )
	
	if self._prioritizedSpaceId then
		logger:infof( "Deactivating space sync prioritization for space %s", self._prioritizedSpaceId )
		self._prioritizedSpaceId = false
		self._prioritizedSpaceIdStream:onNext( false )
	end

end

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

return Rx.Class 'WFCatSpaceFeedsManager'( WFCatSpaceFeedsManager )
