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

NmbLocalCameraProfileStatusWorker.lua

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

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

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

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

	-- wf.core
local KSTableUtils = require 'KSTableUtils'
local WFLogger = require 'WFLogger'
local KSJsonDecoder = require 'KSJsonDecoder'
local WFCloseSupport = require 'WFCloseSupport'
local WFFileUtils = require 'WFFileUtils'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFXmp = require 'WFXmp'
local WFXMLNS = require 'WFXMLNS'
local WFCatFileSpec = require 'WFCatFileSpec'
local NmbProfileCatalogUtils = require 'NmbProfileCatalogUtils'

	-- WF.core.lua.threading
local WFRequestDispatcher = require 'WFRequestDispatcher'

	-- pixel provider
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'

	-- WF.acr
local WFAcrOperations

	-- Rx
local Rx = require "Rx"
local Observable = Rx.Observable
local WFRxArrayDiffer = require 'WFRxArrayDiffer'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'

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

local logger = WFLogger( "NmbLocalCameraProfileStatusWorker" )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt, 
										"KSTableUtils.derefOpt not defined" )

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

local function wrapFnWithSelf( self, fn )

	return function( ... )
		return fn( self, ... )
	end
		
end

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

local function getCameraProfileKey( cameraProfileDigest, model )

	-- Using only digest and model name since the cameraProfileName
	-- sometimes contains the model name, sometimes doesn't.
	return string.lower( string.format( "%s_%s", cameraProfileDigest, model ) )
end

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

local function getMissingProfileDeleteOverlay( localIds )

	local overlays = {}
	for i, localId in ipairs( localIds ) do
		overlays[ #overlays + 1 ] = {
			_id = localId,
			_localOnly = {
				cameraProfileInfo = {
					isMissing = { _delete = true },
				}
			},
		}
	end
	return overlays
end

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

local function isStandardProfileMissing( self, cameraProfileName, cameraModel )

	return self._profileCatalog:getProfileCache():getStandardProfileInfo( cameraProfileName, cameraModel )
		:flatMap( function( standardProfileInfo )
			
			if not standardProfileInfo then
				return Observable.just()
			end
			
			return self._profileCatalog:getProfileCache():getProfilesForModel( cameraModel )
				:map( function( profileInfos )
					
					-- Assuming there are only a few profiles 
					-- per model. 
					for _, info in ipairs( profileInfos or {} ) do
						if info.fileName == standardProfileInfo.fileName then
							local path = self._catalog:fileSpec():pathForStandardProfile( info.relativePath )
							return WFFileUtils.exists( path ) ~= "file"
						end
					end 

					return true

				end )
		end )

end

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

local function isCustomProfileMissing( self, cameraProfileKey )

	local hasLocalCameraProfile = self._hasLocalCameraProfileKeyMap[ cameraProfileKey ]
	if hasLocalCameraProfile then
		-- Profile exists locally
		return Observable.just( false )
	end
	
	if hasLocalCameraProfile == false then
		-- Profile does not exists locally
		return Observable.just( true )
	end
	
	-- We dont have the knowledge of the file yet
	
	return self._cameraProfileInfoMapStream
		:take( 1 )
		:map( function( cameraProfileInfoMap )
			
			local cameraProfileInfo = cameraProfileInfoMap[ cameraProfileKey ]
			
			if not cameraProfileInfo or not cameraProfileInfo.localPath then
				-- Profile doesn't exists locally.
				
				return true
			end
			
			local hasLocalProfile = not not WFFileUtils.exists( cameraProfileInfo.localPath )
			
			-- Update the cache data
			self._hasLocalCameraProfileKeyMap[ cameraProfileKey ] = hasLocalProfile
			
			return not hasLocalProfile
		end )
	
end

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

-- SPAM [ashj 2018-05-09]: 
--[[ As of now, there is no easy way to find if the asset
 	is referring a custom profile and if that is missing.
	Below are the various strategies that we follow to
	determine this.
	
	1. Asset has a profileLink : We will be able to
		find if the info belongs to standard profile or not,
		by looking at the link. If it is not a standard
		profile, and we have the details of this profile
		locally, then we will be able to find the details
		of the reference to the profile.
	
	2. Asset has cameraModel and profileDigest and profileName:
		We will be able to find if it is a standard profile, 
		by using the profileName. 
	
	3. Asset without any information: We need to parse
		the xmp develop settings to find out the actual
		status. We are caching the result so that, next
		time we wont do the parsing again for the same
		asset, unless the settings changes.
		
		If the xmp is externalized, we will do the parsing
		only if the file is locally available. These assets
		will be NOT be considered as missing profile assets.
		We will not cache the data for this, so that next
		time we will retry to check the status of these
		assets.
]]
--------------------------------------------------------------------------------

local function isProfileMissing( self, cameraProfileDigest, cameraProfileName, 
	cameraModel, isStandardProfile )

	if not cameraProfileDigest then
		return Observable.just( false )
	end
	
	if isStandardProfile then
		return isStandardProfileMissing( self, cameraProfileName, cameraModel )
	end
	
	local key = getCameraProfileKey( cameraProfileDigest, cameraModel )
	return isCustomProfileMissing( self, key )

end

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

local function getXmpString( docValue )

	
	local xmpCameraRaw = KSTableUtils_derefOpt( docValue, "develop",
															"xmpCameraRaw" )
															
	if type( xmpCameraRaw ) == 'string' then

		return Rx.Observable.just( xmpCameraRaw )

	elseif type( xmpCameraRaw ) == 'table' and xmpCameraRaw.sha256 then

		logger:infof( "LocalId %s has externalized XMP", docValue._localId )
	
		-- xmp externalized in separate file
		local xmpDevelopInfo = WFCatFileSpec.getXmpDevelopInfoFromDocValue( docValue ) or {}
		local xmpFilePath = xmpDevelopInfo.path

		-- Currently not downloading the externalized file
		-- if it doesn't exists.
		
		if xmpFilePath and WFFileUtils.exists( xmpFilePath ) == 'file' then

			return WFAsyncFileUtils.readFile( xmpFilePath )

				:map( function( fileData )

					if fileData then

						local retVal = tostring( fileData )
						fileData:close()
						return retVal
					
					else
					
						logger:infof( "failed to read external xmp from %s", xmpFilePath )

					end

				end )
				:catch( function( err )
					logger:infof( "Error while reading develop xmp file. Err:%s", err )
					return Rx.Observable.throw( "Error while reading externalized xmp develop file" )
				end )
			
		else
			logger:infof( "UNABLE TO read externalized XMP for localId:%s", docValue._localId )
			return Rx.Observable.throw( "Externalized xmp develop file not present locally" )
		end
	
	end
	
	logger:infof( "xmpCameraRaw of type %s encountered", type( xmpCameraRaw ) )
	return Rx.Observable.just( nil )

end

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

local function getCameraDigestAndName( docValue, xmpCameraRaw )

	if not xmpCameraRaw then
		-- No develop settings
		logger:debugf( "LocalId:%s No develop settings present", docValue._localId )
		return
	end
											
	local wfXmp = WFXmp.WFXmp()

	if not wfXmp:parse( xmpCameraRaw ) then
		-- Parsing failed.
		-- We will retry this in next launch
		logger:errorf( "LocalId:%s xmp parsing failed", docValue._localId )
		return Rx.Observable.throw( "Xmp develop parsing failed for localId:%s", docValue._localId )
	end
	
	
	local cameraProfileDigest = wfXmp:get( WFXMLNS.crs, "CameraProfileDigest" )
	cameraProfileDigest = cameraProfileDigest and cameraProfileDigest:lower()
	
	local cameraProfileName = wfXmp:get( WFXMLNS.crs, "CameraProfile" )
	cameraProfileName = cameraProfileName and cameraProfileName:lower()

	return cameraProfileDigest, cameraProfileName
end

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

local function getProfileInfoFromXmp( self, docValue )
	
	logger:debugf( "LocalId:%s Gonna parse xmp to find profile", docValue._localId )
				
	return getXmpString( docValue )
			
	:flatMap( function( xmpCameraRaw )
	
		local cameraProfileDigest, cameraProfileName =
			getCameraDigestAndName( docValue, xmpCameraRaw )
		
		if not ( cameraProfileDigest and cameraProfileDigest:len() > 0 ) then
			
			-- No valid profile
			-- TODO [ashj 2018-05-10]: Check if this is
			-- true for LrClassic synced images as well.
			-- tknoll had mentioned before that LrClassic
			-- sometimes used to remove the digest.
			logger:debugf( "LocalId:%s. No valid profile", docValue._localId )
			return Observable.just()
		end

		-- Has a valid profile
		-- Check if it is external profile
		local cameraModel = KSTableUtils_derefOpt( docValue, "xmp", "tiff", "Model" )
		
		if NmbProfileCatalogUtils.isInternalProfile( cameraProfileName ) then
			-- This is an embedded profile. No need to store
			-- any info about it.
			return Observable.just()
		end
		
		return self._profileCatalog:getProfileCache():isStandardProfile( cameraProfileName, cameraModel )
			:map( function( isStandardProfile )
				
				logger:debugf( "LocalId:%s Found external profile:%s isStd: %s", 
					docValue._localId, cameraProfileDigest, isStandardProfile )
				
				return cameraProfileDigest, isStandardProfile, cameraProfileName
				
			end )

	end )
end

--------------------------------------------------------------------------------
--[[	
	The assetsNeedingCameraProfileLinks doesn't take keep the profile
	value up to date. Hence we will not be able to rely on that.
	It might be more reliable to parse the figure out the profile
	if we dont have the profile link. Since the parsing happens on
	a worker thread, it should be fine wrt performance too.
	
	Old Thought ->
	
		1. Assets without profile link
			a. Maybe we imported the dcp locally, but offline, so no link.
				i. Assets should be in the assetsNeedingCameraProfileLinks table.
				ii. If not, then we will get a notification very soon with the actual link.
			b. Or no profile 
				i. Don't do anything
			c. Assets with profile, without custom profile links
				i. Edge case, maybe from LrClassic.
				ii. Will not get rendered correctly in oz or locally.
				iii. If LrCC imports it later, then case 1.a  or 2 will apply.
				
	In this, we are handling only case 1.a
]]
local function getProfileInfoForAssetsWithNoLink( self, docValue )

	return getProfileInfoFromXmp( self, docValue )
end

--------------------------------------------------------------------------------
--[[
	2. Assets with profile link
		a. Custom profile link
			i. Custom profile exists.
				1) Get the info from custom profile table and add the localOnly
			ii. Custom profile info doesn't exists locally yet.
				1) Parse the xmp and get info.
				2) The metadata could sync soon, hence we will not have to do this often.
			
		b. Standard profile link
			i. Assume its standard profile for now. If not the link will get updated and we will get a notification with the proper link

]]
--local function getProfileInfoForAssetsWithLink( self, docValue )
--
--	local cameraProfileLink = WFCatFileSpec.getCameraProfileLinkFromDocValue( docValue )
--	if not cameraProfileLink then
--		-- No camera profile link.
--		-- Parse xmp develop settings 
--		return getProfileInfoFromXmp( self, docValue )
--	end
--	
--	local isCustomProfile = isCustomProfileFromLink( self, cameraProfileLink )
--	if not isCustomProfile then
--		-- Case 2.b
--		logger:debugf( "LocalId:%s Link is not of custom profile", 
--				docValue._localId )
--		return Observable.just()
--	end
--	
--	return self._cameraProfileInfoMapStream
--		:take( 1 )
--		:flatMap( function( cameraProfileInfoMap, cameraProfileHrefMap )
--			
--			local customProfileKey = cameraProfileHrefMap[ cameraProfileLink ]
--			if not customProfileKey then
--				
--				-- Case 2.1.ii
--				logger:debugf( "LocalId:%s. Has profile link, but we dont know about it yet.", 
--					docValue._localId )
--				return getProfileInfoFromXmp( self, docValue )
--			end
--			
--			local cameraProfileInfo = cameraProfileInfoMap[ customProfileKey ]
--			return Observable.just( cameraProfileInfo.cameraProfileDigest )
--		end )
--	
--end

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

local function getProfileInfo( self, docValue )

	-- Act as if no link is present in all cases, to cause xmp parsing.

--	local cameraProfileLink = WFCatFileSpec.getCameraProfileLinkFromDocValue( docValue )
--	if not cameraProfileLink then
--		-- No camera profile link.
--		-- Parse xmp develop settings 
		return getProfileInfoForAssetsWithNoLink( self, docValue )
--	end
--	
--	return getProfileInfoForAssetsWithLink( self, docValue )
end

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

local function updateProfileInfoForAsset( self, docValue )

	return getProfileInfo( self, docValue )
		:flatMap( function( cameraProfileDigest, isStandardProfile, cameraProfileName )
			
			if not cameraProfileDigest then
				return Observable.just()
			end
			
			local cameraModel = KSTableUtils_derefOpt( docValue, "xmp", "tiff", "Model" )
			return isProfileMissing( self, cameraProfileDigest, cameraProfileName, 
					cameraModel, isStandardProfile )
				:map( function( isMissing )
					return cameraProfileDigest, isMissing, isStandardProfile
				end )
		end )
			
		:map( function( cameraProfileDigest, isMissing, isStandardProfile )
			
			logger:debugf( "LocalId:%s Received info: digest:%s, isMissing:%s", 
				docValue._localId, cameraProfileDigest, isMissing )
			local xmpCameraRawProfileDigest = NmbProfileCatalogUtils.getXmpCameraRawProfileDigest( docValue )
			
			local overlay = {
				_id = docValue._id,
				_localOnly = {
					cameraProfileInfo = {
						profileDigest = cameraProfileDigest or "NA",
						isMissing = isMissing or { _delete = true },
						isStandard = not not isStandardProfile,
						xmpCameraRawProfileDigest = xmpCameraRawProfileDigest,
					}
				},
			}
			-- SPAM [ashj 2018-05-11]: The asset doc might have
			-- changed from what we were seeing before and we
			-- will be updating a wrong value here. But thats
			-- just temporary. This will go to the indexing again
			-- and if the digest is incorrect, we will again
			-- fix it.
			return overlay
		end )
		:catch( function( err )
			
			-- Errors will get retried on next launch
			
			logger:debugf( "LocalId:%s Error encountered while finding the custom profile:%s", 
				docValue._localId, err )
				
			return Observable.empty()
		end )
end
--------------------------------------------------------------------------------

local function populateAssetCameraProfileStatusStream( self )

	local sql = [[ SELECT docId, cameraProfileDigest, isMissing
		FROM 
			assetReferredCameraProfile 
		WHERE
			cameraProfileDigest IS NULL ]]

	return self._catalog:indexStore():query {
			sql = sql,
			quiesceUntilUpToDate = true,
			calm = 30,
		}
		:map( KSTableUtils.flatten )
		:call( WFRxArrayDiffer )
		:flatMap( function( additions, removals )
			
			-- Process in batch
			return Observable.from( additions )
				:bufferWithCount( 100 )
				:flatMap( function( localIds )
					
					return self._catalog:docStore():bulkDocValue( localIds, nil, "populateAssetCameraProfile" )
				end )
				
				:flatMap( Observable.from )
				:flatMap( wrapFnWithSelf( self, updateProfileInfoForAsset ) )
				:bufferWithCount( 100 )
				:flatMap( function( overlays )
					-- SPAM [ashj 2018-09-24]: There can be multiple
					-- updates to cameraProfileDigest and hence the
					-- addition and removal of the overlay should be
					-- sequential. WFRxThrottledBatchedBulkApplyOverlay
					-- doesn't guarantee that. And hence using normal
					-- bulk overlay api.
					return self._catalog:docStore():bulkApplyOverlay( overlays, "populateAssetCameraProfile" )
				end )
		end )
		
		
end

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

local function nonEmptyArray( array )

	return array and #array > 0
		
end

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

local function isNonNil( v )

	return v ~= nil

end

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

local function prependWith( val )

	return function( ... )

		return val, ...

	end
end

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

local function pluckKey( t, k )

	local result = {}
	for _, v in ipairs( t ) do
		if v and v[ k ] then
			result[ #result + 1 ] = v[ k ]
		end
	end
	return result
end

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

local function getLocalOnlyOverlaysForIds( docIds, localOnlyTable )

	local overlays = {}
	for i, id in ipairs( docIds ) do
		overlays[ i ] = {
			_id = id,
			_localOnly = localOnlyTable,
		}
	end
	
	return overlays
end

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

local function setAssetUpdateDoneStatus( self, docIds )

	-- Mark the Camera profile as asset update done,
	-- so that we dont have to do any operation related
	-- to this profile again.
	
	logger:debugf( "Updating status as done for %s profiles", tostring( #docIds ) )
	
	if #docIds == 0 then
		return Observable.empty()
	end
	
	local profileOverlays = getLocalOnlyOverlaysForIds( docIds, {
		assetUpdateDone = true,
	})
	
	-- Custom profile will be updates only once with the
	-- asset update flag. Hence we can use the throttled
	-- api here.
	return WFRxThrottledBatchedBulkApplyOverlay( "NmbLocalCameraProfileStatusWorker_setAssetUpdateDoneStatus", self._catalog:docStore(), profileOverlays, 100, 10 )
end

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

local function setAssetUpdateDoneStatusObservable( self, docIds )

	return Observable.create( function( observer )
		
		local subscription = setAssetUpdateDoneStatus( self, docIds )
			:subscribe( observer )
		
		return Rx.Disposable.create( function()
			subscription:dispose()
		end )
	end )
end

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

local function removeMissingProfileTag( self, cameraProfileDigestList )

	-- Fix all the assets referring to these profiles
	local sql = [[ SELECT a.docId
		FROM assetReferredCameraProfile AS a
		WHERE a.cameraProfileDigest IN CARRAY( ?, ?, 'char*' ) ]]
	
	return self._catalog:indexStore():queryOnce {
		sql = sql,
		quiesceUntilUpToDate = true,
		cameraProfileDigestList,
		#cameraProfileDigestList
	}
	
	:filter( nonEmptyArray )
	:map( KSTableUtils.flatten )
	:flatMap( Observable.from )
	:bufferWithCount( 100 )
	:flatMap( function( localIds )
		
		-- We need not recheck the profile digest
		-- since if the digest would have changed
		-- we would have received the develop digest
		-- change and we would have cleared that info
		if not localIds or #localIds == 0 then
			return Observable.empty()
		end
		
		-- SPAM [ashj 2018-10-09]: There can be multiple
		-- updates to cameraProfileDigest and hence the
		-- addition and removal of the overlay should be
		-- sequential. WFRxThrottledBatchedBulkApplyOverlay
		-- doesn't guarantee that. And hence using normal
		-- bulk overlay api.
		local assetOverlays = getMissingProfileDeleteOverlay( localIds )
		return self._catalog:docStore():bulkApplyOverlay( assetOverlays )
	end )
	
	-- Let this observable fire only once atmost.
	:toArray()
	:tapOnCompleted( function()
		logger:debugf( "Finished updating assets for %s profiles", tostring( #cameraProfileDigestList ) )
	end )

end

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

local function noteNewLocalCustomCameraProfile( self, newLocalCameraProfileKeyList )
	
	return self._cameraProfileInfoMapStream
		:take( 1 )
		:flatMap( function( cameraProfileInfoMap )
			
			-- Clear the cached value of the new profiles
			for i, key in ipairs( newLocalCameraProfileKeyList ) do
				self._hasLocalCameraProfileKeyMap[ key ] = nil
			end
			
			return Observable.from( newLocalCameraProfileKeyList )
				
				:flatMap( function( newLocalCameraProfileKey )
					return isCustomProfileMissing( self, newLocalCameraProfileKey )
						:map( prependWith( newLocalCameraProfileKey ) )
				end )
				
				-- filter non-missing ones
				:filter( function( newLocalCameraProfileKey, isMissing )
					return not isMissing
				end ) 
				
				:map( function( newLocalCameraProfileKey )
					
					local cameraProfileInfo = cameraProfileInfoMap[ newLocalCameraProfileKey ]
					if not cameraProfileInfoMap[ newLocalCameraProfileKey ] then
						logger:errorf( "Camera profile information is not available: key:%s", newLocalCameraProfileKey )
						return
					end
					
					return cameraProfileInfo
				end )
				
				:filter( isNonNil )
				
				:toArray()
				
				:flatMapLatest( function( cameraProfileInfoList )
				
					local cameraProfileDigestList = pluckKey( cameraProfileInfoList, "cameraProfileDigest" )
					local docIds = pluckKey( cameraProfileInfoList, "docId" )
					
					return removeMissingProfileTag( self, cameraProfileDigestList )
						-- Once the assets are updated, update the profile status.
						:concat( setAssetUpdateDoneStatusObservable( self, docIds ))
					
				end )
		end )

end

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

local function noteNewLocalStdCameraProfiles( self, profilePathList )

	-- TODO [ashj 2018-08-24]: We dont have enough information
	-- to map the filename to a digest now. But once oz adds the
	-- cameraProfileDigest and modelName to the standard profile
	-- assets, it should be possible to find the mapping.
	-- For now, we are fetching the digest from ACR.
	-- Change the implementation once oz adds the digest field.
	
	return Observable.from( profilePathList )
		:flatMap( function( profilePath )
			
			logger:debugf( "Received new std profile:%s", profilePath )
			
			if not WFAcrOperations then
				WFAcrOperations = require 'WFAcrOperations'
			end

			return WFAcrOperations.getCustomCameraProfileInfo( profilePath )
				:map( function( profileInfo )
					local cameraProfileDigest = profileInfo.cameraProfileDigest:lower()
					return cameraProfileDigest
				end )
				:catch( function( errInf )

					-- Not much we can do about that, so carry on after logging.
					logger:errorf( "error occurred while attempting to get info for %s", profilePath )

					return Rx.Observable.empty()
				end )
		end )
		:filter( isNonNil )
		:toArray()
		:flatMap( wrapFnWithSelf( self, removeMissingProfileTag ) )

end

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

local function customCameraProfileStatusStream( self )

	local columnCount = 7
	local sql = [[ SELECT docId, cameraProfileUniqueID, cameraProfileDigest, 
		cameraModel, localPath, href, assetUpdateDone
		FROM 
			customCameraProfiles ]]

	return self._catalog:indexStore():query {
			sql = sql,
			quiesceUntilUpToDate = true,
			calm = 10,
		}
		:map( function( results )
		
			-- Got new results
				
			local cameraProfileInfoMap = {}
			local cameraProfileHrefToKeyMap = {}
			local newProfileKeys = {}
			
			for i, result in ipairs( results ) do
				local docId, _, cameraProfileDigest, cameraModel, localPath, 
					href, assetUpdateDone = unpack( result, 1, columnCount )
				local key = getCameraProfileKey( cameraProfileDigest, 
					cameraModel )
					
				cameraProfileInfoMap[ key ] = {
					docId = docId,
					cameraProfileDigest = cameraProfileDigest,
					cameraModel = cameraModel,
					localPath = localPath,
					href = href,
					assetUpdateDone = assetUpdateDone,
				}
				
				if href then
					cameraProfileHrefToKeyMap[ href ] = key
				end
				
				-- Only new profiles need to be processed.
				local hasLocalPath = KSTableUtils_derefOpt( cameraProfileInfoMap, key, "localPath" )
				local assetUpdatePending = KSTableUtils_derefOpt( cameraProfileInfoMap, key, "assetUpdateDone" ) == 0
				if hasLocalPath and assetUpdatePending then
					newProfileKeys[ #newProfileKeys + 1 ] = key
				end
			end
			
			self._cameraProfileInfoMapStream:onNext( cameraProfileInfoMap, cameraProfileHrefToKeyMap )
			return newProfileKeys
				
		end )
			
		:call( WFRxArrayDiffer )
		
		:tap( function( additions, removals )
		
			-- The assets which are referring to the
			-- changed profiles should get refreshed.
			
			-- We dont support removal of camera profiles
			-- yet. And we need not re-render if the
			-- profile is removed.
			
			if not WFAcrOperations then
				WFAcrOperations = require 'WFAcrOperations'
			end

			if #additions > 0 then
				self._cameraProfileRefreshStream:onNext( additions )
				
				WFAcrOperations.refreshACRDatabases()
					
					:concat( noteNewLocalCustomCameraProfile( self, additions ))
						
					:call( WFRxRunUntilDone )
			end

		end )

end

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

local function getGenericLogFunctionsForSubscribe( streamName )

	return {
		function()
			-- noop onNext fn
		end,
		
		function( err )
			logger:errorf( "Stream: %s ended with error: %s",
									streamName, err )
		end
	}
	
end

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

local NmbLocalCameraProfileStatusWorker = {

	class = {},
	
	_catalog = false,
	_profileCatalog = false,
	
	_cameraProfileRefreshStream = false,
	
	_hasLocalCameraProfileKeyMap = false,
	
	_cameraProfileInfoMapStream = false,

	_assetCameraProfileStatusStream = false,
	
	_continuousTaskForFixingAssetProfileRefs = false,
	
	_listnerSubscription = false,
		-- Composite subscription to manage all subscriptions
	
	WFCloseSupport.addCloseSupportMethods,

}

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

function NmbLocalCameraProfileStatusWorker:__init( catalog, profileCatalog )

	self._catalog = catalog
	self._profileCatalog = profileCatalog
	self._hasLocalCameraProfileKeyMap = {}
	self._assetCameraProfileStatusStream = Rx.ReplaySubject( 1 )
	self._cameraProfileRefreshStream = Rx.ReplaySubject( 1 )
	
	self._cameraProfileInfoMapStream = Rx.ReplaySubject( 1 )
	local cameraProfileInfoMapSubscription = customCameraProfileStatusStream( self )
		:subscribe( unpack( getGenericLogFunctionsForSubscribe( "customCameraProfileStatusStream" ) ) )
	
	self._listnerSubscription = Rx.CompositeDisposable(
		cameraProfileInfoMapSubscription
	)
	
	local taskDelays = {
		progressiveBackoff = false,
		errorDelay = 10,
		maxDelay = 60,
	}
	
	local taskForFixingAssetProfileRefs = function() return populateAssetCameraProfileStatusStream( self ) end
	self._continuousTaskForFixingAssetProfileRefs = WFRxContinuousTask( "AssetProfileReferenceFixerTask", taskForFixingAssetProfileRefs, taskDelays )
	
end

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

function NmbLocalCameraProfileStatusWorker:close()

	if self._listnerSubscription then
		self._listnerSubscription:dispose()
		self._listnerSubscription = nil
	end
	
	if self._continuousTaskForFixingAssetProfileRefs then
		self._continuousTaskForFixingAssetProfileRefs:stopContinuousTask()
		self._continuousTaskForFixingAssetProfileRefs:close()
	end
end


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

function NmbLocalCameraProfileStatusWorker:populateMissingProfileInfo()

	self._continuousTaskForFixingAssetProfileRefs:startContinuousTask()

end

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

function NmbLocalCameraProfileStatusWorker:getCameraProfileRefreshStream()

	return self._cameraProfileRefreshStream

end

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

function NmbLocalCameraProfileStatusWorker:noteNewStandardProfiles( newProfilePathList )

	if #newProfilePathList == 0 then
		return Observable.empty()
	end
	
	if not WFAcrOperations then
		WFAcrOperations = require 'WFAcrOperations'
	end

	return WFAcrOperations.refreshACRDatabases()
		:concat( noteNewLocalStdCameraProfiles( self, newProfilePathList ))
		

end

--------------------------------------------------------------------------------
-- we use this thread so that we can cache profile lookups in a single place
-- and share the lookup info with all calling threads
--------------------------------------------------------------------------------

local function makeProfileCatalog( serverBaseUrl, userId, userInfoCachePath,
								managedCatalogIdOrDoc, rootCatalogDataDir,
								catalogOptions )

	local wfCatUser = require 'WFCatUser'( serverBaseUrl, userId, userInfoCachePath )

	local wfCatSpec = require 'WFCatSpec'( wfCatUser, managedCatalogIdOrDoc )

	return require 'NmbProfileCatalog'( wfCatSpec, rootCatalogDataDir, catalogOptions )

end

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

function NmbLocalCameraProfileStatusWorker.class.initMessageProcessor( 
												Class, 
												encodedPackedWFCatalogArgs )

	local packedWFCatalogArgs = KSJsonDecoder.decodeWebSafeJson(
								encodedPackedWFCatalogArgs )

	local serverBaseUrl, userId, userInfoCachePath,
		managedCatalogIdOrDoc, rootCatalogDataDir,
		catalogOptions,
		maybeProfileCatalogIdOrDoc, maybeProfileCatalogOptions = unpack( packedWFCatalogArgs )
		
	local wfCatalog = PixelProviderSharedTaskUtils.makeWFCatalog( 
		serverBaseUrl, userId, userInfoCachePath,
		managedCatalogIdOrDoc, rootCatalogDataDir, catalogOptions )
		
	local profileCatalog = makeProfileCatalog( 
		serverBaseUrl, userId, userInfoCachePath,
		maybeProfileCatalogIdOrDoc, rootCatalogDataDir,
		maybeProfileCatalogOptions )

	local cameraProfileStatusWorker = Class( wfCatalog, profileCatalog )

	WFCloseSupport.mustCloseAtThreadTermination( 
					cameraProfileStatusWorker )

	WFRequestDispatcher.onMessageDo {

		message = 'populateMissingProfileInfo',

		target = cameraProfileStatusWorker,

	}
	
	WFRequestDispatcher.onMessageDo {

		message = 'getCameraProfileRefreshStream',

		target = cameraProfileStatusWorker,

	}
	
	WFRequestDispatcher.onMessageDo {

		message = 'noteNewStandardProfiles',

		target = cameraProfileStatusWorker,

	}

	WFRequestDispatcher.enableDiagnosticLogForRequestStats()

end

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

return Rx.Class( 'NmbLocalCameraProfileStatusWorker', 
					NmbLocalCameraProfileStatusWorker )
